Unresolved behavior of SQLAlchemy merge during retry scenarios

Answer

The most reliable way to retry write operations without duplicating inserts or losing updates is to use a fresh transaction context—async_session.begin() (or a new synchronous Session) for each retry attempt—combined with explicit clearing of the identity map (expunge or expire) before calling merge. merge alone does not guarantee idempotent behavior across retries, and the autocommit execution option is not uniformly supported across dialects, so it should not be relied upon for retry safety.

Confirmed facts

• merge() returns a persistent instance that is either loaded from the database or created and then has its state copied onto an existing instance present in the Session's identity map.
• After a rollback, if the Session is not cleared, merge may encounter a stale instance in the identity map, treat the object as already persistent, and skip the INSERT, which can result in no new row or an IntegrityError on commit.
• Calling Session.expunge(instance) or Session.expire([instance]) (or beginning a sub‑transaction with begin_nested) removes the stale state so that merge behaves as if the object were new.
• In SQLAlchemy 1.4 and 2.0 the merge implementation uses the unit of work to issue a SELECT for existing rows based on primary key, so the described behavior is consistent across those versions when the identity map is not cleared.

Likely explanation

When a transient failure causes a flush to fail and the transaction is rolled back, the instance may retain a temporary primary‑key value (e.g., None or a negative placeholder). If the same Session is reused and merge is called again without clearing the identity map, SQLAlchemy may see the instance as already associated with the Session and attempt to re‑apply the same INSERT, leading to a duplicate‑key error or a lost update when another session has modified the row in the meantime.

Steps to achieve safe retries

1. At the start of each retry iteration, obtain a fresh transaction context:
   • For async code: async with async_session.begin() as session:
   • For sync code: with Session() as session: (or manually call session.begin()).
2. If you must reuse the same Session object across retries, explicitly clear any instances that participated in the failed flush before the next merge:

   session.expunge_all()   # or session.expunge(instance)
   # or
   session.expire([instance])

3. Call merge on the instance you wish to upsert.
4. Optionally, add a pessimistic lock to protect against concurrent modifications if your dialect supports it:

   session.query(MyModel).filter_by(id=instance.id).with_for_update().one()

   Note: verify that your target dialect (e.g., PostgreSQL, MySQL) supports FOR UPDATE in the desired isolation level.
5. Commit the transaction; if an IntegrityError occurs, retry the loop (the fresh session ensures a clean state).

Missing diagnostic detail

To decide whether to rely on with_for_update or to fall back to database‑level upsert (ON CONFLICT / MERGE), please confirm which SQL dialect you are using (e.g., PostgreSQL, MySQL, SQLite) and its version, as lock behavior and native upsert support vary.