SqlAlchemy unique str field (for use as `sqlalchemy.orm.composite`)

crc_str.py

"""
Author: Rinat Khabibiev <[email protected]>

Compact indexable version of string column for `sqlalchemy` ORM

Uses crc64 hash of string as index value. Supports querying, filtering,
model instantiation and assignment

Usage example:
class User(DeclarativeBase):
    __tablename__ = 'users'
    email = sqlalchemy.orm.composite(
        CrcStr,
        Column('email', String, key='_email'),
        Column('email_crc64', BigInteger, key='_email_crc64', unique=True),
        comparator_factory=CrcStrComparator,
    )
"""
import crcmod.predefined
import operator
import sqlalchemy.ext.mutable
import sqlalchemy.orm.properties
import sqlalchemy.sql.operators
import sqlalchemy.util

make_crc64 = crcmod.predefined.mkPredefinedCrcFun('crc-64')

try:
    str = unicode
except NameError:
    pass


def to_bytes(obj, encoding='utf-8'):
    if isinstance(obj, (bytes, bytearray, memoryview)):
        return bytes(obj)
    if obj is None:
        return b''
    try:
        return obj.__bytes__()
    except AttributeError:
        return str(obj).encode(encoding=encoding)


class CrcStr(str, sqlalchemy.ext.mutable.MutableComposite):

    def __new__(cls, value, crc64=None, *args, **kwargs):
        if isinstance(value, cls):
            return value
        return super().__new__(cls, value, *args, **kwargs)

    def __composite_values__(self):
        return self, self.crc64

    @sqlalchemy.util.memoized_property
    def crc64(self):
        return make_crc64(to_bytes(self))
        
    @classmethod
    def coerce(cls, key, value):
        return cls(value)


class CrcStrComparator(sqlalchemy.orm.properties.CompositeProperty.Comparator):

    @property
    def value(self):
        return self.clauses.clauses[0]

    @property
    def crc64(self):
        return self.clauses.clauses[1]

    def __eq__(self, other):
        return self.crc64 == CrcStr(other).crc64

    def __ne__(self, other):
        return self.crc64 != CrcStr(other).crc64

    def in_(self, other):
        return sqlalchemy.sql.operators.in_op(
            self.crc64,
            map(operator.attrgetter('crc64'), map(CrcStr, other)),
        )

    def notin_(self, other):
        return sqlalchemy.sql.operators.notin_op(
            self.crc64,
            map(operator.attrgetter('crc64'), map(CrcStr, other)),
        )

    def operate(self, op, *other, **kwargs):
        return op(self.value, *other, **kwargs)