如何在字段比较操作中实现值转换，在SqlAlchemy列类型（或另一个类）中？

Question

在不同的表中有许多字段（10+），它们的逻辑需要封装在类型实现中。例如：字符串的上限

import sqlalchemy as sa

class UpperString(sa.String):
    class comparator_factory(sa.String.Comparator):
        def new_sql_operator(self, value: str):
            ...

        # override sql operator
        def match(self, value: str, **kw):
            ...


    def bind_processor(self, dialect):
        """Transform values on db store and query filter parameter on request"""
        def process(value):
            if value is not None:
                value = value.upper()
            return value
        return process

Base = declarative_base(metadata=metadata)

class Name(Base):
    __tablename__ = 'name'
    name = sa.Column('name', UpperString)
    desc = sa.Column('desc', UpperString)

如何在现场比较操作期间实施价值评估

n = Name(name='name')  # new object
assert n.name == 'Name'
n = Name(name='NAME')  # object returning from session.query(Name).filter_by(name='name').one()
assert n.name == 'Name'

与帮助中描述的类似

class CaseInsensitiveName(Comparator):
    ...

class Name(Base):
    ...
    @hybrid_property
    def name(self):
        return CaseInsensitiveName(self._name)

    @name.setter
    def name(self, value):
        self._name = value.upper()

。。。但是没有为每个字段和/或模型创建getter/setter

Answer 1

也许代码不是干净的，而是枯燥的（自定义类型继承的简单实现）和工作

在decorator中实现处理启动器

def operate_comparator(cls):
    class object_processor(Comparator):
        """Transform field value on create model object and parameter on field comparison operations"""
        def operate(self, op, *other, **kwargs):
            other = other[0]
            if not isinstance(other, type(self)):
                other = cls.transform_value(other)
            return op(self.expression, other)

        def __repr__(self):
            return self.expression

    def db_processor(self, dialect=None):
        """Transform field value on db store and query parameter on request"""
        return cls.transform_value

    setattr(cls, 'operate_comparator', object_processor)
    setattr(cls, 'bind_processor', db_processor)
    return cls

在自定义类型类中，您只需要实现一个数据转换器

import sqlalchemy as sa

@operate_comparator
class UpperString(sa.String):
    class comparator_factory(sa.String.Comparator):
        def new_sql_operator(self, value: str):
            ...

    @staticmethod
    def transform_value(value):
        if value is not None:
            value = value.upper()
        return value

创建一个类型类。 重新分配field->_field，对于完全自动创建属性，它不起作用。在setattr之后，类型从InstrumentedAttribute变为propertyProxy。因此，模型类中的列应命名为_field，创建的属性将被命名为field

class Name(Base):
    __tablename__ = 'name'
    _name = sa.Column('name', UpperString)
    _desc = sa.Column('desc', UpperString)

动态创建属性

def make_property_handler(column_name, column_type):
    def get_(self):
        field = getattr(self, column_name)
        return column_type.operate_comparator(field)

    def set_(self, value):
        value = column_type.bind_processor()(value)
        setattr(self, column_name, value)

    def cmp_(cls):
        field = getattr(cls, column_name)
        return column_type.comparator_factory(field)

    return get_, set_, cmp_


def add_property(model, column, column_type):
    property_name = column[1:]
    if hasattr(model, property_name):
        raise AttributeError(f'Class field with name "{property_name}" is present')
    get_, set_, cmp_ = make_property_handler(column, column_type)
    prop = hybrid_property(get_).setter(set_).comparator(cmp_)
    setattr(model, property_name, prop)

…关于我们在现场的活动

@sa.event.listens_for(Base, 'attribute_instrument')
def configure_listener(model, key, inst):
    if hasattr(inst.property, 'columns'):
        column_type = inst.property.columns[0].type
        if hasattr(column_type, 'transform_value') and key.startswith('_'):
            add_property(model, key, column_type)