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

Question

SQLAlchemy 提供PickleType 并为任何可变类型（如字典）提供mutation tracking。

SQLAlchemy 文档提到这是实现可变 PickleType 的方法，但它没有具体说明如何进行。

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

你是如何实现的？

Answer 1

虽然文档中提到了一些示例，但在我看来还不够，所以我将在此处添加我的实现，该实现可用于实现一个可变字典，该字典被腌制并存储在数据库中。

使用文档中的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 会有一个）。

Answer 2

这是我想出的解决方案。它包装任何类型并检测任何属性集并调用 Mutable.changed()。它还包装函数调用并通过拍摄对象前后的快照并进行比较来检测更改。应该适用于 Pickleable 类型...

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 对象时更新它们的另一种方法是在提交更改之前复制和分配它们。