python - query - ¿Cómo puedo verificar los tipos de datos de columna en el ORM de SQLAlchemy?

sqlalchemy relationship not null

Question

Usando el ORM de SQLAlchemy, quiero asegurarme de que los valores sean del tipo correcto para sus columnas.

Por ejemplo, digamos que tengo una columna Integer. Intento insertar el valor "hola", que no es un entero válido. SQLAlchemy me permitirá hacer esto. Solo más tarde, cuando ejecuto session.commit() , sqlalchemy.exc.DataError: (DataError) invalid input syntax integer: "hello"… una excepción: sqlalchemy.exc.DataError: (DataError) invalid input syntax integer: "hello"…

Estoy agregando lotes de registros y no quiero confirmar después de cada add(…) , por razones de rendimiento.

Entonces, ¿cómo puedo:

• Aumente la excepción tan pronto como lo haga session.add(…)
• ¿O asegúrese de que el valor que estoy insertando se pueda convertir al tipo de datos de la columna de destino, antes de agregarlo al lote?
• O cualquier otra forma de evitar que un registro incorrecto arruine a un entero commit() .

Answer 1

Mejorando la respuesta de @zzzeek, ​​sugiero la siguiente solución:

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)

Eso te permite hacer cosas como:

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

Esto tiene dos ventajas: en primer lugar, solo se activa un evento, cuando en realidad hay un validador adjunto al objeto de campo de datos. No desperdicia ciclos de CPU valiosos en eventos set para objetos, que no tienen una función de validación definida. En segundo lugar, le permite definir sus propios tipos de campo y simplemente agregar un método de validación allí, por lo que no todas las cosas que desea almacenar como Integer etc. se ejecutan a través de las mismas comprobaciones, solo las derivadas de su nuevo tipo de campo.

Answer 2

SQLAlchemy no incluye esto ya que difiere a la base de datos DBAPI como la mejor y más eficiente fuente de validación y coacción de valores.

Para crear su propia validación, por lo general se usa la validación de TypeDecorator u ORM. TypeDecorator tiene la ventaja de que opera en el núcleo y puede ser bastante transparente, aunque solo ocurre cuando se emite SQL.

Para hacer la validación y la coerción antes, esto es en el nivel ORM.

La validación puede ser ad-hoc, en la capa ORM, a través de @validates :

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

El sistema de eventos que utiliza @validates también está disponible directamente. Puede escribir una solución generalizada que vincule los validadores de su elección con los tipos que se asignan:

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"

La validación y la coacción también se pueden crear a nivel de tipo usando TypeDecorator, aunque esto solo ocurre cuando se emite SQL, como en este ejemplo que obliga a utilizar cadenas de caracteres utf-8 a unicode:

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