5

Python/SQLAlchemyによって維持されている一連の関連テーブルがあります。特定のテーブルの行を削除する場合、間違いが発生した場合に、将来のある時点でその削除を元に戻すことができるようにする必要があります。is_deleted列を使用してこれを実行し、それをフィルター処理することはできますが、他のテーブルに関連データを照会する場合は、これが面倒になります。is_deleted列を他のすべてのテーブルに追加し、メインテーブルの行が削除されたら、それらすべてを切り替えます。しかし、すべてのテーブルに対するすべてのクエリについて、is_deletedでフィルタリングする必要があります。それは可能ですが、もっと良い戦略があることを望んでいます。

1つの考えは、削除されたすべてのデータを、削除されたデータのみを格納する別のテーブルのセットに移動することです。しかし、SQLAlchemyで特定のオブジェクトが関連付けられているテーブルを切り替えることができるかどうかはわかりません。これが好ましい解決策になると思いますが、それができるかどうかはわかりません。

もう1つの考えは、2番目のデータベースを実行して、削除されたデータをコピーできるということです。しかし、それは私が避けたい管理の複雑さの層を追加します。

どんな考えでもいただければ幸いです。

4

2 に答える 2

5

多くの人が「is_deleted」のことをします。PreFilteredQueryにそのレシピがありますが、私もそのファンではないことに同意します。

他の誰かが提案したように、あなたが探しているのは「バージョン管理」レシピです。SQLAlchemyドキュメントのVersionedObjectsで紹介されている別のバージョン管理されたテーブルにデータのコピーを格納する包括的な例があります。

ここでは、その例で使用されている手法のいくつかを採用して、「削除された」オブジェクトのみを具体的に追跡し、特定の行をメインテーブルに「復元」する「復元」機能を含むより直接的なレシピを作成しました。したがって、「SQLAlchemyを使用すると、特定のオブジェクトが関連付けられているテーブルを切り替えることができます」というよりも、プライマリクラスに似た別のマップされたクラスが作成され、要求に応じて削除を「元に戻す」ために使用できるようになります。 。境界線より下にあるものはすべて、__main__概念実証です。

from sqlalchemy.orm import Session, object_session
from sqlalchemy import event

def preserve_deleted(class_):
    def copy_col(col):
        newcol = col.copy()
        newcol.constraints = set()
        return newcol
    keys = class_.__table__.c.keys()
    cols = dict(
        (col.key, copy_col(col)) for col in class_.__table__.c
    )
    cols['__tablename__'] = "%s_deleted" % class_.__table__.name

    class History(object):
        def restore(self):
            sess = object_session(self)
            sess.delete(self)
            sess.add(copy_inst(self, class_))

    hist_class = type(
                    '%sDeleted' % class_.__name__,
                    (History, Base),
                    cols)

    def copy_inst(fromobj, tocls):
        return tocls(**dict(
                        (key, getattr(fromobj, key))
                        for key in keys
                    ))
    @event.listens_for(Session, 'before_flush')
    def check_deleted(session, flush_context, instances):
        for del_ in session.deleted:
            if isinstance(del_, class_):
                h = copy_inst(del_, hist_class)
                session.add(h)
    class_.deleted = hist_class
    return class_

if __name__ == '__main__':

    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, ForeignKey, Integer, String
    from sqlalchemy.orm import relationship, Session
    from sqlalchemy import create_engine

    Base = declarative_base()

    @preserve_deleted
    class A(Base):
        __tablename__ = "a"

        id = Column(Integer, primary_key=True)
        data1 = Column(String)
        data2 = Column(String)

    @preserve_deleted
    class B(Base):
        __tablename__ = 'b'
        id = Column(Integer, primary_key=True)
        data1 = Column(String)
        a_id = Column(Integer, ForeignKey('a.id'))
        a = relationship("A")

    e = create_engine('sqlite://', echo=True)

    Base.metadata.create_all(e)

    s = Session(e)

    a1, a2, a3, a4 = \
        A(data1='a1d1', data2='a1d2'),\
        A(data1='a2d1', data2='a2d2'),\
        A(data1='a3d1', data2='a3d2'),\
        A(data1='a4d1', data2='a4d2')

    b1, b2, b3, b4 = \
        B(data1='b1', a=a1),\
        B(data1='b2', a=a1),\
        B(data1='b3', a=a3),\
        B(data1='b4', a=a4)

    s.add_all([
        a1, a2, a3, a4,
        b1, b2, b3, b4
    ])
    s.commit()

    assert s.query(A.id).order_by(A.id).all() == [(1, ), (2, ), (3, ), (4, )]
    assert s.query(B.id).order_by(B.id).all() == [(1, ), (2, ), (3, ), (4, )]

    s.delete(a2)
    s.delete(b2)
    s.delete(b3)
    s.delete(a3)
    s.commit()

    assert s.query(A.id).order_by(A.id).all() == [(1, ), (4, )]
    assert s.query(B.id).order_by(B.id).all() == [(1, ), (4, )]

    a2_deleted = s.query(A.deleted).filter(A.deleted.id == 2).one()
    a2_deleted.restore()

    b3_deleted = s.query(B.deleted).filter(B.deleted.id == 3).one()
    a3_deleted = s.query(A.deleted).filter(A.deleted.id == 3).one()
    b3_deleted.restore()
    a3_deleted.restore()

    s.commit()

    assert s.query(A.id).order_by(A.id).all() == [(1, ), (2, ), (3, ), (4, )]
    assert s.query(B.id).order_by(B.id).all() == [(1, ), (3, ), (4, )]
于 2012-11-06T21:41:50.543 に答える
1

django-reversionに似たものを実装しようと思います。

これは、他のテーブルのシリアル化されたデータに加えて、それがどのテーブルであるかなどの情報を含むことができるテーブルがあることを意味します。

例として、django-reversionのVersionモデルを見てください。このcontent_typeフィールドは、モデル情報を含むDjangoモデルを参照します。この場合、テーブル名を含むcharフィールドである可能性があります（ただし、テーブルの数が多い場合は、コンテンツタイプのテーブルの方が適しています）。

次に、コードを追加して、挿入または更新するたびに、バージョンテーブルも更新されるようにします。何かを回復したいときはいつでも、バージョンテーブルからシリアル化されたデータを取得し、レコードを再挿入するだけです。

M2M、カスケード削除など、いくつかの注意点があるかもしれません。しかし、私はそこから始めます。

于 2012-11-06T21:07:56.667 に答える