python - 変更時に自動的に更新される可変PickleTypesを実装する方法

Question

SQLAlchemyは、（dictのように）可変であるすべてのタイプの突然変異追跡PickleTypeを提供します。

SQLAlchemyのドキュメントには、これがミュータブルを実装する方法であると記載されていますPickleTypeが、それをどのように進めるかについては正確に述べられていません。

注：dictをに格納したいと思いPickleTypeます。

これをどのように実装しますか？

Answer 1

ドキュメントにはいくつかの例が記載されていますが、私の目には十分ではないため、ここに実装を追加します。これを使用して、データベースにピクルス化されて保存される可変dictを実装できます。

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オブジェクトを変更したときに確実に更新されるようにする別の方法は、変更をコミットする前にオブジェクトをコピーして割り当てることです。