Python中Sqlalchemy自定义字段类型的Mutation Tracking问题分享

JsonEncodeDict-demo-1.py

import json

https://segmentfault.com/a/1190000004288061


from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.mutable import Mutable
from sqlalchemy.orm import sessionmaker
from sqlalchemy.types import TypeDecorator, VARCHAR


***************************


engine = create_engine(‘sqlite:///./cnblogblog.db’, echo=False)
Base = declarative_base()
DBSession = sessionmaker(bind=engine)


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):
    "Detect dictionary set events and emit change events."

    dict.__setitem__(self, key, value)
    self.changed()

def __delitem__(self, key):
    "Detect dictionary del events and emit change events."

    dict.__delitem__(self, key)
    self.changed()

class MyDataClass1(Base):
tablename = ‘my_data1’
id = Column(Integer, primary_key=True)
data = Column(MutableDict.as_mutable(JSONEncodedDict))
name = Column(String(50))


if name == ‘main’:


Base.metadata.create_all(engine)
session = DBSession()

m1 = MyDataClass1(data={'value1': 'foo1'}, name='xiaohong')
session.add(m1)
session.commit()

#######session 提交后，data 可以关联到 query

m1.name = 'xiaolang'

m1.data['value1'] = 'bar'#数据库的值，将再改变

# assert m1 in session.dirty
session.commit()

# my_data= session.query(MyDataClass).filter_by(id=1).one()
# a= my_data.data
# print (type(a))
# print (a)
# print (a["value1"])
#
# my_data.data["value1"] = "foo2"
#
# session.commit()

JsonEncodeDict-demo-2.py

import json

from sqlalchemy import Column, Integer, String


https://segmentfault.com/a/1190000004288061


from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.types import TypeDecorator, VARCHAR


***************************


engine = create_engine(‘sqlite:///./cnblogblog.db’, echo=False)
Base = declarative_base()
DBSession = sessionmaker(bind=engine)


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):


“Detect dictionary set events and emit change events.”




dict.setitem(self, key, value)


self.changed()




def delitem(self, key):


“Detect dictionary del events and emit change events.”




dict.delitem(self, key)


self.changed()


class MyDataClass2(Base):
tablename = ‘my_data2’
id = Column(Integer, primary_key=True)
data = Column(JSONEncodedDict)
name = Column(String(50))


if name == ‘main’:
Base.metadata.create_all(engine)
session = DBSession()


m1 = MyDataClass2(data={'value1': 'foo1'}, name='xiaohong')
session.add(m1)
session.commit()

m1.name = 'xiaolang'
#######session 提交后，name 关联到了 query

# assert m1 in session.dirty

#######session 提交后，data 将不再关联到 query(解决方法，按着 JsonEncodeDict-demo-1.py 来）

m1.data['value1'] = 'bar'#数据库的值，不再改变
session.commit()
# assert m1 in session.dirty

# my_data= session.query(MyDataClass).filter_by(id=1).one()
# a= my_data.data
# print (type(a))
# print (a)
# print (a["value1"])
#
# my_data.data["value1"] = "foo2"
#
# session.commit()
```


Python中Sqlalchemy自定义字段类型的Mutation Tracking问题分享

1 回复

我最近在项目里也遇到了类似的问题，折腾了半天才搞明白。

核心问题是：当你继承TypeDecorator创建自定义字段类型时，默认情况下SQLAlchemy不会自动追踪这个字段内部值的"变化"。比如你定义了一个JSONEncodedDict类型，直接修改字典内容（像obj.data['key'] = 'value'）是不会触发更新语句的。

解决方案：需要实现TypeDecorator.coerce_compared_value()方法，并返回self.impl.coerce_compared_value(op, value)。更关键的是要设置impl = MutableDict.as_mutable(JSONEncodedDict)，让SQLAlchemy知道这是个可变类型。

from sqlalchemy import TypeDecorator, JSON
from sqlalchemy.ext.mutable import MutableDict
import json

class JSONEncodedDict(TypeDecorator):
    impl = JSON
    
    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
    
    def coerce_compared_value(self, op, value):
        return self.impl.coerce_compared_value(op, value)

# 关键：用MutableDict包装
JSONEncodedDict = MutableDict.as_mutable(JSONEncodedDict)

# 使用
from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

class MyModel(Base):
    __tablename__ = 'my_table'
    id = Column(Integer, primary_key=True)
    data = Column(JSONEncodedDict)

# 现在修改字典内容会自动触发更新
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    obj = MyModel(data={'key': 'old_value'})
    session.add(obj)
    session.commit()
    
    # 直接修改字典 - 现在能正常追踪了
    obj.data['key'] = 'new_value'
    session.commit()  # 会生成UPDATE语句

简单说就是记得用MutableDict.as_mutable()包装你的自定义类型。

回到顶部