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

Question

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

SQLAlchemy文档提到这是实现可变的PickleType的方式，但它没有详细说明如何继续处理它。

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

你如何实现这个？

Answer 1

虽然文档中提到了一些示例，但在我眼中这还不够，所以我将在此处添加我的实现，以用于实现在数据库中进行pickle和存储的可变字典。

从文档使用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（）。它还封装了函数调用，并通过拍摄前后对象的快照并进行比较来检测更改。应适用于与pickle类型...

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