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

Question

我在使用SQLAlchemy这个工具时，想确保我放入数据库的值是正确的类型，跟它们对应的列匹配。

举个例子，假设我有一个整型的列。我试着插入一个值“hello”，这显然不是一个有效的整数。SQLAlchemy会让我这样做，但只有在我执行session.commit()的时候，它才会报错：sqlalchemy.exc.DataError: (DataError) invalid input syntax integer: "hello"…。

我一次性添加很多记录，不想每添加一条就提交，这样效率太低了。

所以我想知道，我该怎么做：

• 能不能在我执行session.add(…)的时候就立刻报错？
• 或者，在把值放进批量之前，确保我插入的值可以转换成目标列的数据类型？
• 或者有没有其他办法，防止一个错误的记录影响到整个commit()的操作？

Answer 1

在@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")

这个方法有两个好处：首先，只有在数据字段对象上确实有验证器时，才会触发事件。这样就不会浪费宝贵的CPU资源去处理那些没有定义验证功能的对象的set事件。其次，它允许你定义自己的字段类型，并在其中添加验证方法，这样并不是所有你想存储为Integer的东西都要经过相同的检查，只有那些来自你新字段类型的才会。

Answer 2

SQLAlchemy并没有内置这个功能，因为它认为数据库API（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