Module DINO2.types

Custom types

# -*- coding: utf-8 -*-
"""Custom types"""

from __future__ import annotations

from datetime import date, timedelta
from enum import Enum
from sqlalchemy import String, Integer, cast, func
from sqlalchemy.engine.interfaces import Dialect
from sqlalchemy.types import TypeDecorator, TypeEngine
from typing import Optional, Type, Tuple, Any, Union


# https://stackoverflow.com/q/35209650
class DinoDate(TypeDecorator):
    """Column type for date strings in yyyymmdd format, converted to `datetime.date` objects"""
    impl = String(length=8)

    cache_ok = True

    def coerce_compared_value(self, op, value):
        if isinstance(value, str):
            return self.impl
        return self

    class comparator_factory(TypeDecorator.Comparator, impl.comparator_factory):
        @property
        def year(self):
            return cast(func.substr(self.expr, 1, 4), Integer)

        @property
        def month(self):
            return cast(func.substr(self.expr, 5, 2), Integer)

        @property
        def day(self):
            return cast(func.substr(self.expr, 7, 2), Integer)

    def process_bind_param(self, value: Optional[date], dialect: Optional[Dialect] = None) -> Optional[str]:
        if value is None:
            return None
        if isinstance(value, date):
            return '%04d%02d%02d' % (value.year, value.month, value.day)
        raise ValueError(f"expected datetime.date, got {value.__class__.__name__}")

    def process_result_value(self, value: Optional[str], dialect: Optional[Dialect] = None) -> Optional[date]:
        if value:
            return date(year=int(value[0:4]), month=int(value[4:6]), day=int(value[6:8]))
        return None


class DinoTimeDelta(TypeDecorator):
    """Column type for timedeltas saved as int in the database, converted to `datetime.timedelta` objects"""
    impl = Integer
    should_evaluate_none = True

    def __init__(self, minus_1_none: bool = False):
        super().__init__()
        self.minus_1_none = minus_1_none

    # do not coerce None to "is NULL" automatically
    coerce_to_is_types: Tuple[Type, ...] = tuple()

    def coerce_compared_value(self, op, value):
        if isinstance(value, int):
            return self.impl
        return self

    class comparator_factory(TypeDecorator.Comparator, impl.comparator_factory):
        @property
        def total_seconds(self):
            return cast(self.expr, Integer)

    def process_bind_param(self, value: Optional[timedelta], dialect: Optional[Dialect] = None) -> Optional[int]:
        if value is None:
            return -1 if self.minus_1_none else None
        if isinstance(value, timedelta):
            return int(value.total_seconds())
        raise ValueError(f"expected datetime.timedelta, got {value.__class__.__name__}")

    def process_result_value(self, value: Optional[int], dialect: Optional[Dialect] = None) -> Optional[timedelta]:
        if (value is None) or (value == -1 and self.minus_1_none):
            return None
        return timedelta(seconds=int(value))

    def copy(self, **kwargs):
        return DinoTimeDelta(self.minus_1_none)


# https://stackoverflow.com/a/38786737
class TypeEnum(TypeDecorator):
    """Column type for python `enum.Enum`s stored in the database"""
    impl: Union[TypeEngine, Type[TypeEngine]] = TypeEngine

    def __init__(self, enum_type: Type[Enum]):
        if self.__class__ == TypeEnum:
            raise NotImplementedError(f"Do not use {self.__class__} directly")
        super().__init__()
        self.enum_type = enum_type

    def process_bind_param(self, value: Optional[Enum], dialect: Optional[Dialect] = None) -> Optional[Any]:
        if value is None:
            return None
        if isinstance(value, self.enum_type):
            return value.value
        raise ValueError(f'expected {self.enum_type.__name__} value, got {value.__class__.__name__}')

    def process_result_value(self, value: Optional[Any], dialect: Optional[Dialect] = None) -> Optional[Enum]:
        if value is None:
            return None
        return self.enum_type(value)

    def copy(self, **kwargs):
        return self.__class__(self.enum_type)


class IntEnum(TypeEnum):
    """Column type for python `enum.Enum`s stored as int in the database"""
    impl = Integer

    def coerce_compared_value(self, op, value):
        if isinstance(value, int):
            return self.impl
        return self


class CharEnum(TypeEnum):
    """Column type for python `enum.Enum`s stored as a single character in the database"""
    impl = String(length=1)

    def coerce_compared_value(self, op, value):
        if isinstance(value, str):
            return self.impl
        return self

Classes

class DinoDate (*args, **kwargs)

Column type for date strings in yyyymmdd format, converted to datetime.date objects

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Expand source code

class DinoDate(TypeDecorator):
    """Column type for date strings in yyyymmdd format, converted to `datetime.date` objects"""
    impl = String(length=8)

    cache_ok = True

    def coerce_compared_value(self, op, value):
        if isinstance(value, str):
            return self.impl
        return self

    class comparator_factory(TypeDecorator.Comparator, impl.comparator_factory):
        @property
        def year(self):
            return cast(func.substr(self.expr, 1, 4), Integer)

        @property
        def month(self):
            return cast(func.substr(self.expr, 5, 2), Integer)

        @property
        def day(self):
            return cast(func.substr(self.expr, 7, 2), Integer)

    def process_bind_param(self, value: Optional[date], dialect: Optional[Dialect] = None) -> Optional[str]:
        if value is None:
            return None
        if isinstance(value, date):
            return '%04d%02d%02d' % (value.year, value.month, value.day)
        raise ValueError(f"expected datetime.date, got {value.__class__.__name__}")

    def process_result_value(self, value: Optional[str], dialect: Optional[Dialect] = None) -> Optional[date]:
        if value:
            return date(year=int(value[0:4]), month=int(value[4:6]), day=int(value[6:8]))
        return None

Ancestors

• sqlalchemy.sql.type_api.TypeDecorator
• sqlalchemy.sql.base.SchemaEventTarget
• sqlalchemy.sql.type_api.TypeEngine
• sqlalchemy.sql.visitors.Traversible

Class variables

var impl

var cache_ok

var comparator_factory

A :class:.TypeEngine.Comparator that is specific to :class:.TypeDecorator.

User-defined :class:.TypeDecorator classes should not typically need to modify this.

Methods

def coerce_compared_value(self, op, value)

Suggest a type for a 'coerced' Python value in an expression.

By default, returns self. This method is called by the expression system when an object using this type is on the left or right side of an expression against a plain Python object which does not yet have a SQLAlchemy type assigned::

expr = table.c.somecolumn + 35

Where above, if somecolumn uses this type, this method will be called with the value operator.add and 35. The return value is whatever SQLAlchemy type should be used for 35 for this particular operation.

Expand source code

def coerce_compared_value(self, op, value):
    if isinstance(value, str):
        return self.impl
    return self

def process_bind_param(self, value: Optional[date], dialect: Optional[Dialect] = None) ‑> Optional[str]

Receive a bound parameter value to be converted.

Subclasses override this method to return the value that should be passed along to the underlying :class:.TypeEngine object, and from there to the DBAPI execute() method.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

This operation should be designed with the reverse operation in mind, which would be the process_result_value method of this class.

:param value: Data to operate upon, of any type expected by this method in the subclass. Can be None. :param dialect: the :class:.Dialect in use.

Expand source code

def process_bind_param(self, value: Optional[date], dialect: Optional[Dialect] = None) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, date):
        return '%04d%02d%02d' % (value.year, value.month, value.day)
    raise ValueError(f"expected datetime.date, got {value.__class__.__name__}")

def process_result_value(self, value: Optional[str], dialect: Optional[Dialect] = None) ‑> Optional[datetime.date]

Receive a result-row column value to be converted.

Subclasses should implement this method to operate on data fetched from the database.

Subclasses override this method to return the value that should be passed back to the application, given a value that is already processed by the underlying :class:.TypeEngine object, originally from the DBAPI cursor method fetchone() or similar.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

:param value: Data to operate upon, of any type expected by this method in the subclass. Can be None. :param dialect: the :class:.Dialect in use.

This operation should be designed to be reversible by the "process_bind_param" method of this class.

Expand source code

def process_result_value(self, value: Optional[str], dialect: Optional[Dialect] = None) -> Optional[date]:
    if value:
        return date(year=int(value[0:4]), month=int(value[4:6]), day=int(value[6:8]))
    return None

class DinoTimeDelta (minus_1_none: bool = False)

Column type for timedeltas saved as int in the database, converted to datetime.timedelta objects

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Expand source code

class DinoTimeDelta(TypeDecorator):
    """Column type for timedeltas saved as int in the database, converted to `datetime.timedelta` objects"""
    impl = Integer
    should_evaluate_none = True

    def __init__(self, minus_1_none: bool = False):
        super().__init__()
        self.minus_1_none = minus_1_none

    # do not coerce None to "is NULL" automatically
    coerce_to_is_types: Tuple[Type, ...] = tuple()

    def coerce_compared_value(self, op, value):
        if isinstance(value, int):
            return self.impl
        return self

    class comparator_factory(TypeDecorator.Comparator, impl.comparator_factory):
        @property
        def total_seconds(self):
            return cast(self.expr, Integer)

    def process_bind_param(self, value: Optional[timedelta], dialect: Optional[Dialect] = None) -> Optional[int]:
        if value is None:
            return -1 if self.minus_1_none else None
        if isinstance(value, timedelta):
            return int(value.total_seconds())
        raise ValueError(f"expected datetime.timedelta, got {value.__class__.__name__}")

    def process_result_value(self, value: Optional[int], dialect: Optional[Dialect] = None) -> Optional[timedelta]:
        if (value is None) or (value == -1 and self.minus_1_none):
            return None
        return timedelta(seconds=int(value))

    def copy(self, **kwargs):
        return DinoTimeDelta(self.minus_1_none)

Ancestors

• sqlalchemy.sql.type_api.TypeDecorator
• sqlalchemy.sql.base.SchemaEventTarget
• sqlalchemy.sql.type_api.TypeEngine
• sqlalchemy.sql.visitors.Traversible

Class variables

var impl

A type for int integers.

var should_evaluate_none

var coerce_to_is_types : Tuple[Type, ...]

var comparator_factory

A :class:.TypeEngine.Comparator that is specific to :class:.TypeDecorator.

User-defined :class:.TypeDecorator classes should not typically need to modify this.

Methods

def coerce_compared_value(self, op, value)

Suggest a type for a 'coerced' Python value in an expression.

By default, returns self. This method is called by the expression system when an object using this type is on the left or right side of an expression against a plain Python object which does not yet have a SQLAlchemy type assigned::

expr = table.c.somecolumn + 35

Where above, if somecolumn uses this type, this method will be called with the value operator.add and 35. The return value is whatever SQLAlchemy type should be used for 35 for this particular operation.

Expand source code

def coerce_compared_value(self, op, value):
    if isinstance(value, int):
        return self.impl
    return self

def process_bind_param(self, value: Optional[timedelta], dialect: Optional[Dialect] = None) ‑> Optional[int]

Receive a bound parameter value to be converted.

Subclasses override this method to return the value that should be passed along to the underlying :class:.TypeEngine object, and from there to the DBAPI execute() method.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

This operation should be designed with the reverse operation in mind, which would be the process_result_value method of this class.

:param value: Data to operate upon, of any type expected by this method in the subclass. Can be None. :param dialect: the :class:.Dialect in use.

Expand source code

def process_bind_param(self, value: Optional[timedelta], dialect: Optional[Dialect] = None) -> Optional[int]:
    if value is None:
        return -1 if self.minus_1_none else None
    if isinstance(value, timedelta):
        return int(value.total_seconds())
    raise ValueError(f"expected datetime.timedelta, got {value.__class__.__name__}")

def process_result_value(self, value: Optional[int], dialect: Optional[Dialect] = None) ‑> Optional[datetime.timedelta]

Receive a result-row column value to be converted.

Subclasses should implement this method to operate on data fetched from the database.

Subclasses override this method to return the value that should be passed back to the application, given a value that is already processed by the underlying :class:.TypeEngine object, originally from the DBAPI cursor method fetchone() or similar.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

:param value: Data to operate upon, of any type expected by this method in the subclass. Can be None. :param dialect: the :class:.Dialect in use.

This operation should be designed to be reversible by the "process_bind_param" method of this class.

Expand source code

def process_result_value(self, value: Optional[int], dialect: Optional[Dialect] = None) -> Optional[timedelta]:
    if (value is None) or (value == -1 and self.minus_1_none):
        return None
    return timedelta(seconds=int(value))

def copy(self, **kwargs)

Produce a copy of this :class:.TypeDecorator instance.

This is a shallow copy and is provided to fulfill part of the :class:.TypeEngine contract. It usually does not need to be overridden unless the user-defined :class:.TypeDecorator has local state that should be deep-copied.

Expand source code

def copy(self, **kwargs):
    return DinoTimeDelta(self.minus_1_none)

class TypeEnum (enum_type: Type[Enum])

Column type for python enum.Enums stored in the database

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Expand source code

class TypeEnum(TypeDecorator):
    """Column type for python `enum.Enum`s stored in the database"""
    impl: Union[TypeEngine, Type[TypeEngine]] = TypeEngine

    def __init__(self, enum_type: Type[Enum]):
        if self.__class__ == TypeEnum:
            raise NotImplementedError(f"Do not use {self.__class__} directly")
        super().__init__()
        self.enum_type = enum_type

    def process_bind_param(self, value: Optional[Enum], dialect: Optional[Dialect] = None) -> Optional[Any]:
        if value is None:
            return None
        if isinstance(value, self.enum_type):
            return value.value
        raise ValueError(f'expected {self.enum_type.__name__} value, got {value.__class__.__name__}')

    def process_result_value(self, value: Optional[Any], dialect: Optional[Dialect] = None) -> Optional[Enum]:
        if value is None:
            return None
        return self.enum_type(value)

    def copy(self, **kwargs):
        return self.__class__(self.enum_type)

Ancestors

• sqlalchemy.sql.type_api.TypeDecorator
• sqlalchemy.sql.base.SchemaEventTarget
• sqlalchemy.sql.type_api.TypeEngine
• sqlalchemy.sql.visitors.Traversible

Subclasses

• CharEnum
• IntEnum

Class variables

var impl : Union[sqlalchemy.sql.type_api.TypeEngine, Type[sqlalchemy.sql.type_api.TypeEngine]]

The ultimate base class for all SQL datatypes.

Common subclasses of :class:.TypeEngine include :class:.String, :class:.Integer, and :class:.Boolean.

For an overview of the SQLAlchemy typing system, see :ref:types_toplevel.

Seealso

:ref:types_toplevel

Methods

def process_bind_param(self, value: Optional[Enum], dialect: Optional[Dialect] = None) ‑> Optional[Any]

Receive a bound parameter value to be converted.

Subclasses override this method to return the value that should be passed along to the underlying :class:.TypeEngine object, and from there to the DBAPI execute() method.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

This operation should be designed with the reverse operation in mind, which would be the process_result_value method of this class.

:param value: Data to operate upon, of any type expected by this method in the subclass. Can be None. :param dialect: the :class:.Dialect in use.

Expand source code

def process_bind_param(self, value: Optional[Enum], dialect: Optional[Dialect] = None) -> Optional[Any]:
    if value is None:
        return None
    if isinstance(value, self.enum_type):
        return value.value
    raise ValueError(f'expected {self.enum_type.__name__} value, got {value.__class__.__name__}')

def process_result_value(self, value: Optional[Any], dialect: Optional[Dialect] = None) ‑> Optional[enum.Enum]

Receive a result-row column value to be converted.

Subclasses should implement this method to operate on data fetched from the database.

Subclasses override this method to return the value that should be passed back to the application, given a value that is already processed by the underlying :class:.TypeEngine object, originally from the DBAPI cursor method fetchone() or similar.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

:param value: Data to operate upon, of any type expected by this method in the subclass. Can be None. :param dialect: the :class:.Dialect in use.

This operation should be designed to be reversible by the "process_bind_param" method of this class.

Expand source code

def process_result_value(self, value: Optional[Any], dialect: Optional[Dialect] = None) -> Optional[Enum]:
    if value is None:
        return None
    return self.enum_type(value)

def copy(self, **kwargs)

Produce a copy of this :class:.TypeDecorator instance.

This is a shallow copy and is provided to fulfill part of the :class:.TypeEngine contract. It usually does not need to be overridden unless the user-defined :class:.TypeDecorator has local state that should be deep-copied.

Expand source code

def copy(self, **kwargs):
    return self.__class__(self.enum_type)

class IntEnum (enum_type: Type[Enum])

Column type for python enum.Enums stored as int in the database

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Expand source code

class IntEnum(TypeEnum):
    """Column type for python `enum.Enum`s stored as int in the database"""
    impl = Integer

    def coerce_compared_value(self, op, value):
        if isinstance(value, int):
            return self.impl
        return self

Ancestors

• TypeEnum
• sqlalchemy.sql.type_api.TypeDecorator
• sqlalchemy.sql.base.SchemaEventTarget
• sqlalchemy.sql.type_api.TypeEngine
• sqlalchemy.sql.visitors.Traversible

Methods

def coerce_compared_value(self, op, value)

Suggest a type for a 'coerced' Python value in an expression.

By default, returns self. This method is called by the expression system when an object using this type is on the left or right side of an expression against a plain Python object which does not yet have a SQLAlchemy type assigned::

expr = table.c.somecolumn + 35

Where above, if somecolumn uses this type, this method will be called with the value operator.add and 35. The return value is whatever SQLAlchemy type should be used for 35 for this particular operation.

Expand source code

def coerce_compared_value(self, op, value):
    if isinstance(value, int):
        return self.impl
    return self

Inherited members

• TypeEnum:
  • copy
  • impl
  • process_bind_param
  • process_result_value

class CharEnum (enum_type: Type[Enum])

Column type for python enum.Enums stored as a single character in the database

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Expand source code

class CharEnum(TypeEnum):
    """Column type for python `enum.Enum`s stored as a single character in the database"""
    impl = String(length=1)

    def coerce_compared_value(self, op, value):
        if isinstance(value, str):
            return self.impl
        return self

Ancestors

• TypeEnum
• sqlalchemy.sql.type_api.TypeDecorator
• sqlalchemy.sql.base.SchemaEventTarget
• sqlalchemy.sql.type_api.TypeEngine
• sqlalchemy.sql.visitors.Traversible

Methods

def coerce_compared_value(self, op, value)

Suggest a type for a 'coerced' Python value in an expression.

By default, returns self. This method is called by the expression system when an object using this type is on the left or right side of an expression against a plain Python object which does not yet have a SQLAlchemy type assigned::

expr = table.c.somecolumn + 35

Where above, if somecolumn uses this type, this method will be called with the value operator.add and 35. The return value is whatever SQLAlchemy type should be used for 35 for this particular operation.

Expand source code

def coerce_compared_value(self, op, value):
    if isinstance(value, str):
        return self.impl
    return self

Inherited members

• TypeEnum:
  • copy
  • impl
  • process_bind_param
  • process_result_value