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

Question

SQLAlchemy为任何可变类型（如字典）提供PickleType并提供突变跟踪。

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