sqlalchemy orm數據類型驗證方法比較

1.在定義ORM模型時校驗

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

 1  from sqlalchemy.orm import validates
 2 
 3 class EmailAddress(Base):
 4     __tablename__ = 'address'
 5 
 6     id = Column(Integer, primary_key=True)
 7     email = Column(String)
 8 
 9     @validates('email')
10     def validate_email(self, key, address):
11         assert '@' in address
12         return address

View Code

 2.全局校驗

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

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

獲取orm信息

 1     @classmethod
 2     def orm_fields_info(cls):
 3         """
 4         從ORM中獲取字段信息
 5         :param cls:當前資源類
 6         :type cls:類實例
 7         :returns: 返回ORM字段類型,字段是否可空
 8         """
 9         fields_type = {}
10         fields_nullable = {}
11         orm_meta = cls.orm_meta
12         for prop_name in orm_meta.__dict__:
13             prop = getattr(orm_meta, prop_name)
14             if isinstance(prop, attributes.InstrumentedAttribute):
15                 prop = prop.prop
16                 if isinstance(prop, properties.ColumnProperty):
17                     fields_type[prop_name] = prop.columns[0].type
18                     fields_nullable[prop_name] = prop.columns[0].nullable
19         return fields_type, fields_nullable

View Code

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

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

View Code

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

 1 from sqlalchemy import Column, Integer, String, DateTime
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy import event
 4 import datetime
 5 
 6 Base= declarative_base()
 7 
 8 def validate_int(value):
 9     if isinstance(value, basestring):
10         value = int(value)
11     else:
12         assert isinstance(value, int)
13     return value
14 
15 def validate_string(value):
16     assert isinstance(value, basestring)
17     return value
18 
19 def validate_datetime(value):
20     assert isinstance(value, datetime.datetime)
21     return value
22 
23 validators = {
24     Integer:validate_int,
25     String:validate_string,
26     DateTime:validate_datetime,
27 }
28 
29 # this event is called whenever an attribute
30 # on a class is instrumented
31 @event.listens_for(Base, 'attribute_instrument')
32 def configure_listener(class_, key, inst):
33     if not hasattr(inst.property, 'columns'):
34         return
35     # this event is called whenever a "set" 
36     # occurs on that instrumented attribute
37     @event.listens_for(inst, "set", retval=True)
38     def set_(instance, value, oldvalue, initiator):
39         validator = validators.get(inst.property.columns[0].type.__class__)
40         if validator:
41             return validator(value)
42         else:
43             return value
44 
45 
46 class MyObject(Base):
47     __tablename__ = 'mytable'
48 
49     id = Column(Integer, primary_key=True)
50     svalue = Column(String)
51     ivalue = Column(Integer)
52     dvalue = Column(DateTime)
53 
54 
55 m = MyObject()
56 m.svalue = "ASdf"
57 
58 m.ivalue = "45"
59 
60 m.dvalue = "not a date"

View Code