python sqlalchemy 使用Dict字段

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