Source code for fictive.transform.ext.sqlalchemy

"""
Integrating Transform with SQLAlchemy

"""

import argparse
from functools import cached_property

import sqlalchemy

from fictive.transform import Transform


class TransformColumnType(sqlalchemy.types.TypeDecorator):
    """
    A TypeDecorator (i.e., SQLAlchemy column type) that automatically applies a transform

    """
    impl = sqlalchemy.types.TypeEngine

    def __init__(
            self, base_type, transform, *args, pass_null=True, **kwargs):
        if not isinstance(transform, Transform):
            raise TypeError('`transform` must be a `Transform` object')

        self.__base_type = base_type
        self.__base_type_args = argparse.Namespace(args=args, kwargs=kwargs)
        self.__transform = transform
        self.__pass_null = pass_null

        super().__init__(*args, **kwargs)

    @cached_property
    def __impl(self):
        if isinstance(self.__base_type, type):
            return self.__base_type(
                *self.__base_type_args.args, **self.__base_type_args.kwargs)
        return self.__base_type

    def load_dialect_impl(self, dialect):
        return self.__impl

    @property
    def python_type(self):  # pragma: no cover
        raise NotImplementedError()

    def process_bind_param(self, value, dialect):
        """
        reverse transform of external value to database value

        """
        if value is None and self.__pass_null:
            return None
        return self.__transform.inverse_operation(value)

    def process_result_value(self, value, dialect):
        """
        transform database value to external value

        """
        if value is None and self.__pass_null:
            return None
        return self.__transform(value)

    def process_literal_param(self, value, dialect):  # pragma: no cover
        return self.process_bind_param(value, dialect)