SqlAlchemy对应的数据库的表结构,数据库是没有Dict字段的,所以只能通过json
来存储。
而想要将其做成自动存储,需要使用sqlalchemy.ext.mutable
模块。
sqlalchemy_type.py
import json
from sqlalchemy.types import TypeDecorator, VARCHAR
from sqlalchemy.ext.mutable import Mutable
class JSONEncodedDict(TypeDecorator):
"Represents an immutable structure as a json-encoded string."
impl = VARCHAR
def process_bind_param(self, value, dialect):
if value is not None:
value = json.dumps(value)
return value
def process_result_value(self, value, dialect):
if value is not None:
value = json.loads(value)
return value
class MutableDict(Mutable, dict):
@classmethod
def coerce(cls, key, value):
"Convert plain dictionaries to MutableDict."
if not isinstance(value, MutableDict):
if isinstance(value, dict):
return MutableDict(value)
# this call will raise ValueError
return Mutable.coerce(key, value)
else:
return value
def __setitem__(self, key, value):
"""捕获设置事件,并将自己设置为已修改"""
dict.__setitem__(self, key, value)
self.changed()
def __delitem__(self, key):
"""捕获删除事件,并将自己设置为已修改"""
dict.__delitem__(self, key)
self.changed()
def update(self, *args, **kwargs):
"""捕获update事件,并将自己设置为已修改"""
dict.update(self, *args, **kwargs)
self.changed()
使用经典映射
orm.py
from sqlalchemy import Table, Column, Integer
from sqlalchemy_type import MutableDict, JSONEncodedDict
my_data = Table('my_data', metadata,
Column('id', Integer, primary_key=True),
Column('data', MutableDict.as_mutable(JSONEncodedDict))
)
或者使用关系映射
orm.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy_type import MutableDict, JSONEncodedDict
Base = declarative_base()
class MyDataClass(Base):
__tablename__ = 'my_data'
id = Column(Integer, primary_key=True)
data = Column(MutableDict.as_mutable(JSONEncodedDict))
转载请注明来源,欢迎对文章中的引用来源进行考证,欢迎指出任何有错误或不够清晰的表达。可以在下面评论区评论,也可以邮件至 365433079@qq.com