sqlalchemy orm數據類型校驗的幾種方法

1.在定義ORM模型時校驗

sqlalchemy提供validates函數支持對字段的校驗

from sqlalchemy.orm import validates

class EmailAddress(Base):
    __tablename__ = 'address'

    id = Column(Integer, primary_key=True)
    email = Column(String)

    @validates('email')
    def validate_email(self, key, address):
        assert '@' in address
        return address

 

2.全局校驗

①.根據sqlalchemy數據類型對應的python_type(注:有些數據類型沒有實現python_type這個方法,需重寫)

在進行增改.commit()之前,對傳入數據校驗.字段的數據類型及是否為空可從已定義好的orm model中獲取.

獲取orm信息

@classmethod
    def orm_fields_info(cls):
        """
        從ORM中獲取字段信息
        :param cls:當前資源類
        :type cls:類實例
        :returns: 返回ORM字段類型,字段是否可空
        """
        fields_type = {}
        fields_nullable = {}
        orm_meta = cls.orm_meta
        for prop_name in orm_meta.__dict__:
            prop = getattr(orm_meta, prop_name)
            if isinstance(prop, attributes.InstrumentedAttribute):
                prop = prop.prop
                if isinstance(prop, properties.ColumnProperty):
                    fields_type[prop_name] = prop.columns[0].type
                    fields_nullable[prop_name] = prop.columns[0].nullable
        return fields_type, fields_nullable

 

根據orm 字段類型的python_type對傳入的數據類型進行校驗

 

def validate_data_type(orm_fields_info, field_name, field_value):
    fields_type, fields_nullable = orm_fields_info
    field_column = fields_type.get(field_name, None)
    if field_column:
        if not (not field_value and fields_nullable[field_name]):
            if hasattr(field_column, 'python_type'):
                if not isinstance(field_value, field_column.python_type):
                    raise exceptions.ValidationError(attribute=field_name_display,
                         msg=_(
                             "field [%s] data type didn't match! require [%s],found type [%s]"
                             % (field_name, field_column.python_type, type(field_value))))

 

②.利用sqlalchemy event創建通用校驗器

 

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"