Как я могу проверить типы данных Column в SQLAlchemy ORM?

Question

Используя SQLAlchemy ORM, я хочу убедиться, что значения имеют правильный тип для своих столбцов.

Например, скажем, у меня есть столбец целых чисел. Я пытаюсь вставить значение «привет», которое не является допустимым целым числом. SQLAlchemy позволит мне сделать это. Только позже, когда я выполняю session.commit(), возникает ли исключение: sqlalchemy.exc.DataError: (DataError) invalid input syntax integer: "hello"….

Я добавляю пакеты записей и не хочу фиксировать после каждого add(…) по соображениям производительности.

Так как я могу:

• Поднять исключение, как только я session.add(…)
• Или убедитесь, что добавляемое мной значение можно преобразовать в целевой тип данных Столбец, за до добавления его в пакет?
• Или любым другим способом, чтобы одна испорченная запись не испортила весь commit().

Answer 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

Answer 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 для объектов, для которых не определена функция проверки.Во-вторых, он позволяет вам определять свои собственные типы полей и просто добавлять туда метод валидатора, поэтому не все вещи, которые вы хотите сохранить как Integer и т. Д., Проходят те же проверки, только те, которые получены из вашего нового типа поля.