SqlAlchemy 標記字段變更

來自：http://graycarl.me/post/notes/sqlalchemy-set-column-modified

為了在數據庫里面存儲一些比較隨意的結構化數據，我參照(抄襲)sqlalchemy的文檔實現了一個自定義類型：

class JSON(TypeDecorator):
    """ Json String Field """
    impl = Text

    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif isinstance(value, (dict, list)):
            return json.dumps(value)
        else:
            raise ValueError("unsupported value type")

    def process_result_value(self, value, dialect):
        if value is None:
            return value
        else:
            return json.loads(value)


class MyModel(Base):
    ...
    json = Column(JSON, default=[])

    def add_item(self, item):
        self.json.append(item)

其實這本質上就是把python的dict或list直接dumps成json內容存入數據庫，從數據庫讀出之后再自動loads成dict或者list來使用。

但是再使用的過程中發現了一個問題。對這個MyModel的實例無論怎么修改其json字段都無法正常保存到數據庫中，目測應該是SA沒有檢測到這個字段的變化而導致的。重新翻看文檔，發現文檔里面已經有說明了。

Note that the ORM by default will not detect “mutability” on such a type - meaning, in-place changes to values will not be detected and will not be flushed. Without further steps, you instead would need to replace the existing value with a new one on each parent object to detect changes. Note that there’s nothing wrong with this, as many applications may not require that the values are ever mutated once created. For those which do have this requirement, support for mutability is best applied using the sqlalchemy.ext.mutable extension - see the example in Mutation Tracking.

再看看sqlalchemy.ext.mutable模塊的文檔，發現使用起來也沒有那么簡練，而我只想找個方法來主動標記字段變更就行了。于是翻看mutable模塊的代碼，找到了其最重要的一句：

from sqlalchemy.orm.attributes import flag_modified

class Mutable(MutableBase):

    def changed(self):
        """Subclasses should call this method whenever change events occur."""

        for parent, key in self._parents.items():
            flag_modified(parent, key)

其實就是這個flag_modified標記了修改狀態，所以只需要把我自己的代碼改一句就行了。

from sqlalchemy.orm.attributes import flag_modified

class MyModel(Base):
    ...
    json = Column(JSON, default=[])

    def add_item(self, item):
        self.json.append(item)
        flag_modified(self, "json")

打完收工。