如何实现可变的PickleTypes并在更改时自动更新

Question

SQLAlchemy 提供了一种叫做 PickleType 的类型，并且对于任何可变类型（比如字典）都支持 变更追踪。

SQLAlchemy 的文档提到这是实现可变 PickleType 的方法，但并没有详细说明具体该怎么做。

注意: 我想把一个字典存储在 PickleType 中。

那么，应该怎么实现呢？

Answer 1

这是我想到的一个解决方案。它可以包装任何类型，并检测任何属性的变化，然后调用Mutable.changed()。它还可以包装函数调用，通过在调用前后拍摄对象的快照来检测变化，并进行比较。应该适用于可以被序列化的类型...

from sqlalchemy.ext.mutable import Mutable

class MutableTypeWrapper(Mutable):
    top_attributes = ['_underlying_object',
                      '_underlying_type',
                      '_last_state', 
                      '_snapshot_update', 
                      '_snapshot_changed', 
                      '_notify_if_changed',
                      'changed',
                      '__getstate__',
                      '__setstate__',
                      'coerce']

    @classmethod
    def coerce(cls, key, value):
        if not isinstance(value, MutableTypeWrapper):
            try:
                return MutableTypeWrapper(value)
            except:
                return Mutable.coerce(key, value)
        else:
            return value

    def __getstate__(self):
        return self._underlying_object

    def __setstate__(self, state):
        self._underlying_type = type(state)
        self._underlying_object = state

    def __init__(self, underlying_object, underlying_type=None):
        if (underlying_object is None and underlying_type is None):  
            print('Both underlying object and type are none.')
            raise RuntimeError('Unable to create MutableTypeWrapper with no underlying object or type.')

        if (underlying_object is not None):
            self._underlying_object = underlying_object
        else:
            self._underlying_object = underlying_type()

        if (underlying_type is not None):
            self._underlying_type = underlying_type
        else:
            self._underlying_type = type(underlying_object)

    def __getattr__(self, attr):
        if (attr in MutableTypeWrapper.top_attributes):
            return object.__getattribute__(self, attr)

        orig_attr = self._underlying_object.__getattribute__(attr)
        if callable(orig_attr):
            def hooked(*args, **kwargs):
                self._snapshot_update()
                result = orig_attr(*args, **kwargs)
                self._notify_if_changed()
                # prevent underlying from becoming unwrapped
                if result == self._underlying_object:
                    return self
                return result
            return hooked
        else:
            return orig_attr

    def __setattr__(self, attr, value):
        if (attr in MutableTypeWrapper.top_attributes):
            object.__setattr__(self, attr, value)
            return

        self._underlying_object.__setattr__(attr, value)

        self.changed()

    def _snapshot_update(self):
        self._last_state = pickle.dumps(self._underlying_object,
                                        pickle.HIGHEST_PROTOCOL)

    def _snapshot_changed(self):
        return self._last_state != pickle.dumps(self._underlying_object,
                                                pickle.HIGHEST_PROTOCOL)

    def _notify_if_changed(self):
        if (self._snapshot_changed()):
            self.changed()

然后可以像下面这样与PickleType一起使用：

class TestModel(Base):
    __tablename__ = 'testtable'

    id = Column(Integer, primary_key=True)
    obj = Column(MutableTypeWrapper.as_mutable(PickleType))

这里的缺点是，每次函数调用之前，底层类都会拍摄快照，然后在调用后比较变化，以确认底层对象是否发生了变化。这会对性能产生显著影响。

确保你的PickleType对象在修改时更新的另一种方法是，在提交更改之前先复制并赋值。

Answer 2

虽然文档里提到了一些例子，但我觉得还不够，所以我在这里分享我的实现方法，这个方法可以用来创建一个可变的字典，并且可以被序列化后存储到数据库中。

可以参考文档中的MutableDict示例：

class MutableDict(Mutable, dict):

    @classmethod
    def coerce(cls, key, value):
        if not isinstance(value, MutableDict):
            if isinstance(value, dict):
                return MutableDict(value)
            return Mutable.coerce(key, value)
        else:
            return value

    def __delitem(self, key):
        dict.__delitem__(self, key)
        self.changed()

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

    def __getstate__(self):
        return dict(self)

    def __setstate__(self, state):
        self.update(self)

现在创建一个需要跟踪的列：

class MyModel(Base):
    data = Column(MutableDict.as_mutable(PickleType))

我希望能看到一些其他的例子，可能更复杂一些，或者使用不同的数据结构。对于pickle来说，通用的方法是什么样的呢？有没有这样的例子（我想应该没有，否则SQLAlchemy早就有了）。