如何验证 SQLAlchemy ORM 中的列数据类型？答案

【问题标题】：How can I verify Column data types in the SQLAlchemy ORM?如何验证 SQLAlchemy ORM 中的列数据类型？
【发布时间】：2012-02-17 08:09:16
【问题描述】：

使用 SQLAlchemy ORM，我想确保值是其列的正确类型。

例如，假设我有一个整数列。我尝试插入值“hello”，它不是一个有效的整数。 SQLAlchemy 将允许我这样做。只是稍后，当我执行session.commit() 时，它会引发异常：sqlalchemy.exc.DataError: (DataError) invalid input syntax integer: "hello"…。

我正在添加批量记录，出于性能原因，我不想在每个add(…) 之后提交。

那我该怎么办：

我一做就提出异常session.add(…)
或者，确保我插入的值可以转换为目标 Column 数据类型，在将其添加到批处理中？
或任何其他方式来防止一个不良记录破坏整个commit()。

【问题讨论】：

可能重复：stackoverflow.com/questions/2390753/…
@greut 我之前没有看到过这个问题，但评分最高的答案使用了一种已弃用的技术。给出的另一个答案是哲学答案。在这里，我们有 SQLAlchemy 的作者给出了准确而有用的答案。

标签： python database orm sqlalchemy

【解决方案1】：

SQLAlchemy 没有内置此功能，因为它将 DBAPI/数据库视为验证和强制值的最佳和最有效的来源。

要构建自己的验证，通常使用 TypeDecorator 或 ORM 级别的验证。 TypeDecorator 的优点是它在核心运行并且可以非常透明，尽管它只在实际发出 SQL 时发生。

为了更快地进行验证和强制，这是在 ORM 级别。

验证可以是临时的，在 ORM 层，通过@validates：

http://docs.sqlalchemy.org/en/latest/orm/mapped_attributes.html#simple-validators

@validates 使用的事件系统也可以直接使用。您可以编写一个通用的解决方案，将您选择的验证器链接到被映射的类型：

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event
import datetime

Base= declarative_base()

def validate_int(value):
    if isinstance(value, basestring):
        value = int(value)
    else:
        assert isinstance(value, int)
    return value

def validate_string(value):
    assert isinstance(value, basestring)
    return value

def validate_datetime(value):
    assert isinstance(value, datetime.datetime)
    return value

validators = {
    Integer:validate_int,
    String:validate_string,
    DateTime:validate_datetime,
}

# this event is called whenever an attribute
# on a class is instrumented
@event.listens_for(Base, 'attribute_instrument')
def configure_listener(class_, key, inst):
    if not hasattr(inst.property, 'columns'):
        return
    # this event is called whenever a "set" 
    # occurs on that instrumented attribute
    @event.listens_for(inst, "set", retval=True)
    def set_(instance, value, oldvalue, initiator):
        validator = validators.get(inst.property.columns[0].type.__class__)
        if validator:
            return validator(value)
        else:
            return value


class MyObject(Base):
    __tablename__ = 'mytable'

    id = Column(Integer, primary_key=True)
    svalue = Column(String)
    ivalue = Column(Integer)
    dvalue = Column(DateTime)


m = MyObject()
m.svalue = "ASdf"

m.ivalue = "45"

m.dvalue = "not a date"

验证和强制也可以使用 TypeDecorator 在类型级别构建，但这仅在发出 SQL 时，例如将 utf-8 字符串强制为 unicode 的示例：

http://docs.sqlalchemy.org/en/latest/core/custom_types.html#coercing-encoded-strings-to-unicode

【讨论】：

谢谢zzzeek。我以前没有使用过事件系统。很好的例子，展示了在哪里以及如何挂钩。
这个解决方案应该更加宣传。似乎是满足广泛验证要求的最合理方式。
当我想要更严格的输入强制执行时，我发现这对 SQLite 非常有帮助。

【解决方案2】：

改进@zzzeek 的答案，我建议以下解决方案：

from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.event import listen_for

Base = declarative_base()

@listens_for(Base, 'attribute_instrument')
def configure_listener(table_cls, attr, col_inst):
    if not hasattr(col_inst.property, 'columns'):
        return
    validator = getattr(col_inst.property.columns[0].type, 'validator', None)
    if validator:
        # Only decorate columns, that need to be decorated
        @listens_for(col_inst, "set", retval=True)
        def set_(instance, value, oldvalue, initiator):
            return validator(value)

这可以让你做以下事情：

class Name(String):
    def validator(self, name):
        if isinstance(name, str):
            return name.upper()
        raise TypeError("name must be a string")

这有两个好处：首先，只有一个事件被触发，当实际上有一个验证器附加到数据字段对象时。它不会在对象的set 事件上浪费宝贵的 CPU 周期，这些对象没有定义验证功能。其次，它允许您定义自己的字段类型并在其中添加一个验证器方法，因此并非所有要存储为 Integer 等的东西都经过相同的检查，只有从您的新字段类型派生的那些。

【讨论】：