SQLAlchemy Cheatsheet 06 — Transactions and Concurrency

async with AsyncSessionLocal() as session:
    user = User(...)
    session.add(user)
    await session.commit()    # tx commits
    
    await session.get(User, 1)  # new tx starts

async with AsyncSessionLocal() as session:
    async with session.begin():
        session.add(user1)
        session.add(user2)
    # auto-commit on exit; rollback on exception

async with session.begin():
    session.add(parent)
    
    async with session.begin_nested() as sp:
        session.add(child1)
        if oh_no:
            await sp.rollback()
    
    session.add(child2)
# parent + child2 committed; child1 not

# Per engine
engine = create_async_engine(URL, isolation_level="REPEATABLE READ")

# Per connection
await session.connection(execution_options={"isolation_level": "SERIALIZABLE"})

account = await session.scalar(
    select(Account).where(Account.id == 1).with_for_update()
)
account.balance -= 100
await session.commit()

# Error if locked
.with_for_update(nowait=True)

# Skip locked rows (queue pattern)
job = await session.scalar(
    select(Job)
    .where(Job.status == "pending")
    .order_by(Job.created_at)
    .limit(1)
    .with_for_update(skip_locked=True)
)

.with_for_update(read=True)   # FOR SHARE; blocks writers only

.with_for_update(of=Account)   # only lock Account rows

class Post(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    version: Mapped[int] = mapped_column(default=0)
    
    __mapper_args__ = {"version_id_col": "version"}

# SQLAlchemy adds WHERE version = ? to UPDATE
# StaleDataError if no rows match

import asyncio
from sqlalchemy.exc import DBAPIError

async def with_retry(fn, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return await fn()
        except DBAPIError as e:
            sqlstate = getattr(e.orig, "sqlstate", None) or getattr(e.orig, "pgcode", None)
            if sqlstate != "40P01" or attempt == max_attempts - 1:   # 40P01 = deadlock (PG)
                raise
            await asyncio.sleep(0.1 * (2 ** attempt))

await session.connection(execution_options={"postgresql_readonly": True})

await session.connection(execution_options={"isolation_level": "AUTOCOMMIT"})

@event.listens_for(Session, "after_commit")
def on_commit(session): ...

@event.listens_for(Session, "after_rollback")
def on_rollback(session): ...