Non puoi davvero farlo con la Session
da fuori. Session
dovrebbe supportarlo internamente. Implicherebbe il salvataggio di un sacco di stato privato, quindi potrebbe non valere la pena.
Ho completamente abbandonato la maggior parte delle cose ORM a favore dell'interfaccia SQLAlchemy Core di livello inferiore. Usando quella (o anche qualsiasi interfaccia dbapi) puoi usare banalmente il tuo retry_on_deadlock_decorator
decoratore (vedi domanda sopra) per creare un db.execute
che riconosca i tentativi involucro.
@retry_on_deadlock_decorator
def deadlock_safe_execute(db, stmt, *args, **kw):
return db.execute(stmt, *args, **kw)
E invece di
db.execute("UPDATE users SET active=0")
lo fai
deadlock_safe_execute(db, "UPDATE users SET active=0")
che riproverà automaticamente se si verifica un deadlock.