如何使用输入提示简化 SQLAlchemy 2.0 模型的数据验证例程？

Question

假设我有一个代表汽车的 SQLAlchemy 2.0 模型。它包含一个简单的字符串属性验证例程（参考）：

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import validates


class Base(DeclarativeBase):
    pass


class Car(Base):
    __tablename__ = 'car'
    id: Mapped[int] = mapped_column(primary_key=True)
    make: Mapped[str] = mapped_column()  # Tesla
    model: Mapped[str] = mapped_column()  # S
    color: Mapped[str] = mapped_column()  # Black


    # Ensure that the input data is not null
    @validates('make', 'model', 'color')  # <============
    def validate_string(self, key, value):
        if not value:
            raise ValueError(f'The {key} must be entered')
        return value

现在，让我们考虑一下我需要向模型添加 30 个额外的字符串属性并验证它们。鉴于当前的验证例程设置，我必须在验证装饰器中添加 30 个附加属性名称：

    @validates('make', 'model', 'color', 'str_attr_4', 'str_attr_5', 'str_attr_6', ..., 'str_attr_33')

我发现这非常乏味，但我确信有一种更明智的方法可以继续。

我心中有一个解决方案，但我正在努力实现它：由于通过

Mapped

中提供的键入提示，我们已经知道每个属性的类型，如果可以实现以下功能，那就太好了：

class Car(Base):
    __tablename__ = 'car'
    id: Mapped[int] = mapped_column(primary_key=True)
    make: Mapped[str] = mapped_column()  # Tesla
    model: Mapped[str] = mapped_column()  # S
    color: Mapped[str] = mapped_column()  # Black
    str_attr_4: Mapped[str] = mapped_column()
    str_attr_5: Mapped[str] = mapped_column()
    str_attr_6: Mapped[str] = mapped_column()
    ...
    str_attr_33: Mapped[str] = mapped_column()


    # Ensure that the input data is not null
    @validates(*[attr_name for attr_name in dir(self) if attr_name is mapped to str])  # <============
    def validate_string(self, key, value):
        if not value:
            raise ValueError(f'The {key} must be entered')
        return value

我进行了多次尝试和大量研究来构造

*[attr_name for attr_name in dir(self) if attr_name is mapped to str]

表达式，但我无法让它发挥作用。

您对如何实现这一目标有什么建议，或者还有其他简化验证例程的想法吗？

Answer 1

这是可能的，但实施起来很棘手。 TLDR 是您需要在定义

Base

之后但模型之前添加此事件。

import typing

import sqlalchemy as sa
from sqlalchemy import orm
...

class Base(orm.DeclarativeBase):
    pass


@sa.event.listens_for(Base, 'instrument_class', propagate=True)
def receive_instrument_class(mapper: orm.Mapper, class_) -> None:
    # Select the class (or classes) which require the validator.
    if class_.__name__ == 'Car':
        # Find columns of type str.
        string_columns = [
            k for k, v in class_.__annotations__.items() 
            if isinstance(v, typing._GenericAlias) and v.__args__ == (str,)
        ]
        # Apply the validates decorator to the validation method.
        class_.validate_string = orm.validates(*string_columns)(class_.validate_string)

请注意，

orm.validates

仅适用于已明确提供值的情况：如果在构建模型时省略了值，则

validates

将不会被调用，因此对于问题中的用例，它可能不是正确的选择。考虑按照建议在系统边界进行验证。

validates

装饰器的工作原理是将其参数存储为装饰方法上的属性。在对类进行检测后，要验证的属性名称和验证方法将存储在映射器的

validators

属性中 - 这就是为什么我们必须在“仪器类”侦听器中应用装饰器：在映射过程中的任何后期并且不会有任何影响。

在映射过程的这一点上，映射器或模型类的

__dict__

中没有定义任何列。然而，模型类的

__annotations__

属性将列名称映射到类型提示，因此我们可以使用它来标识与

Mapped[str]

关联的列（更复杂的类型声明留给读者作为练习）。

（可能有更好的方法来处理类型声明 - 我看过这个Q&A，但看起来没有稳定的方法来做到这一点，但打字不是我的领域很熟悉）。

这是一个完整的示例：

import typing
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import Mapped, mapped_column


class Base(orm.DeclarativeBase):
    pass


@sa.event.listens_for(Base, 'instrument_class', propagate=True)
def receive_instrument_class(mapper: orm.Mapper, class_) -> None:
    print('Instrument class')
    if class_.__name__ == 'Car':
        # Find columns of type str.
        string_columns = [
            k for k, v in class_.__annotations__.items() 
            if isinstance(v, typing._GenericAlias) and v.__args__ == (str,)
        ]
        # Apply the validates decorator to the validation method.
        class_.validate_string = orm.validates(*string_columns)(class_.validate_string)


class Car(Base):
    __tablename__ = 'car'
    id: Mapped[int] = mapped_column(primary_key=True)
    make: Mapped[str]
    model: Mapped[str]
    color: Mapped[str]
    str_attr_4: Mapped[str]
    str_attr_5: Mapped[str]
    str_attr_6: Mapped[str]
    ...
    str_attr_33: Mapped[str]

    def validate_string(self, key, value):
        if '🤡' in value:
            raise ValueError(f'Value {repr(value)} for {key} is not allowed' )
        return value


engine = sa.create_engine('sqlite://', echo=True)
Base.metadata.create_all(engine)
Session = orm.sessionmaker(engine)

with Session.begin() as s:
    car = Car(make='Tesla', model='S', color='Black', str_attr_6='El🤡n')
    s.add(car)