SQLAlchemy ORM永続化された属性のイベントフック

Question

属性が更新されてデータベースに保持されると、外部APIを呼び出すSQLAlchemyイベントの方法を見つけるために取り組んでいます。ここに私の文脈があります：

Userbirthdayという名前のモデルです。 Userモデルのインスタンスが更新されて保存されると、外部APIを呼び出してこのユーザーの誕生日を適切に更新する必要があります。

私は、しかし、それはあまりにも多くのヒットを生成し、set/remove属性イベントが時にロールバックされます（自動がFalseとトランザクションに設定されているコミット最終的に持続になるだろうことを保証する方法がない、Attribute Events試してみましたエラーが発生しました）。

Session Eventsは、Session/SessionFactoryをパラメータとして必要とするため、セッションが使用されたことに基づいてコード内の場所が多すぎるため、機能しません。

私は公式のドキュメントで可能な限りSQLAlchemy ORM event hooksを見てきましたが、私の要件を満たすものは見つかりませんでした。

SQLAlchemyでこの種の組み合わせイベントトリガーを実装する方法について他の誰かが洞察しているのだろうかと思います。ありがとう。

Answer 1

これは、複数のイベントを組み合わせて行うことができます。あなたが使用する必要がある特定のイベントは、特定のアプリケーションに依存するが、基本的な考え方はこれです：

1. [InstanceEvents.load]インスタンスがロードされたとき、それがロードされ、以降のセッションに追加されなかったという事実を書き留め（インスタンスがロードされている場合のみ初期状態を保存したい）属性が変更された場合は、変更されたこと、必要であれば変更されたことを書き留めておきます（これらの最初の2つのステップ
2. [SessionEvents.before_flush]フラッシュが発生した場合、実際にどのインスタンスが保存されているかをメモしてください。
3. [SessionEvents.before_commit]コミット完了後に[SessionEvents.after_commit]
4. （あなたはそれへのアクセス権を持っていない可能性があるので、もうコミット後）のインスタンスの現在の状態を書き留め、完了をコミットする前に、カスタムイベントハンドラをオフに解雇し、保存したインスタンスをクリアする

興味深い課題は、イベントの順序です。 session.flush()を実行せずにsession.commit()を実行すると、before_flushイベントの前にbefore_commitイベントが発生することがわかります。このイベントは、の前にsession.flush()を実行するシナリオとは異なります。解決方法はbefore_commitコールでsession.flush()に電話して注文を強制することです。これはおそらく100％コーシャーではありませんが、実際には私にとってはうまくいきます。ここで

は、イベントの順序付けの（簡単な）の図です：あなたが見ることができるように

begin 
load 
(save initial state) 
set attribute 
... 
flush 
set attribute 
... 
flush 
... 
(save modified state) 
commit 
(fire off "object saved and changed" event) 

完全な例

from itertools import chain 
from weakref import WeakKeyDictionary, WeakSet 
from sqlalchemy import Column, String, Integer, create_engine 
from sqlalchemy import event 
from sqlalchemy.orm import sessionmaker, object_session 
from sqlalchemy.ext.declarative import declarative_base 

Base = declarative_base() 

engine = create_engine("sqlite://") 
Session = sessionmaker(bind=engine) 


class User(Base): 
    __tablename__ = "users" 

    id = Column(Integer, primary_key=True) 
    birthday = Column(String) 


@event.listens_for(User.birthday, "set", active_history=True) 
def _record_initial_state(target, value, old, initiator): 
    session = object_session(target) 
    if session is None: 
     return 
    if target not in session.info.get("loaded_instances", set()): 
     return 
    initial_state = session.info.setdefault("initial_state", WeakKeyDictionary()) 
    # this is where you save the entire object's state, not necessarily just the birthday attribute 
    initial_state.setdefault(target, old) 


@event.listens_for(User, "load") 
def _record_loaded_instances_on_load(target, context): 
    session = object_session(target) 
    loaded_instances = session.info.setdefault("loaded_instances", WeakSet()) 
    loaded_instances.add(target) 


@event.listens_for(Session, "before_flush") 
def track_instances_before_flush(session, context, instances): 
    modified_instances = session.info.setdefault("modified_instances", WeakSet()) 
    for obj in chain(session.new, session.dirty): 
     if session.is_modified(obj) and isinstance(obj, User): 
      modified_instances.add(obj) 


@event.listens_for(Session, "before_commit") 
def set_pending_changes_before_commit(session): 
    session.flush() # IMPORTANT 
    initial_state = session.info.get("initial_state", {}) 
    modified_instances = session.info.get("modified_instances", set()) 
    del session.info["modified_instances"] 
    pending_changes = session.info["pending_changes"] = [] 
    for obj in modified_instances: 
     initial = initial_state.get(obj) 
     current = obj.birthday 
     pending_changes.append({ 
      "initial": initial, 
      "current": current, 
     }) 
     initial_state[obj] = current 


@event.listens_for(Session, "after_commit") 
def after_commit(session): 
    pending_changes = session.info.get("pending_changes", {}) 
    del session.info["pending_changes"] 
    for changes in pending_changes: 
     print(changes) # this is where you would fire your custom event 

    loaded_instances = session.info["loaded_instances"] = WeakSet() 
    for v in session.identity_map.values(): 
     if isinstance(v, User): 
      loaded_instances.add(v) 


def main(): 
    engine = create_engine("sqlite://", echo=False) 
    Base.metadata.create_all(bind=engine) 
    session = Session(bind=engine) 

    user = User(birthday="foo") 
    session.add(user) 
    user.birthday = "bar" 
    session.flush() 
    user.birthday = "baz" 
    session.commit() # prints: {"initial": None, "current": "baz"} 
    user.birthday = "foobar" 
    session.commit() # prints: {"initial": "baz", "current": "foobar"} 

    session.close() 


if __name__ == "__main__": 
    main() 

、それは少し複雑で、非常に人間工学的ではありません。それがORMに統合されていればより良いでしょうが、そうしない理由もあるかもしれません。