SQLAlchemy + Postgres Cheatsheet 07 — Locking, SKIP LOCKED, Advisory Locks

Cheatsheet for Postgres locking. Long-form: Postgres textbook Ch 5 .

SELECT FOR UPDATE

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

Locks the row until commit. Other tx wait.

NOWAIT

account = await session.scalar(
    select(Account).where(Account.id == 1).with_for_update(nowait=True)
)

Errors immediately if locked (instead of waiting).

SKIP LOCKED (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)
)
if job:
    job.status = "processing"
    await session.commit()

Each worker grabs an unlocked row. Others skip locked.

.with_for_update(read=True)

Blocks writers but allows multiple readers to hold the lock.

OF (lock specific tables in joins)

.with_for_update(of=Account)

In a multi-table join, only lock Account rows.

Advisory locks (transaction-scoped)

await session.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": 12345})
# ... critical section ...
await session.commit()   # auto-releases

Advisory locks (session-scoped)

await session.execute(text("SELECT pg_advisory_lock(:key)"), {"key": 12345})
# ... work ...
await session.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": 12345})

Released only on explicit unlock or session close. Be careful with connection pools.

Two-int advisory locks (namespaced)

await session.execute(
    text("SELECT pg_advisory_xact_lock(:k1, :k2)"),
    {"k1": LOCK_NAMESPACE, "k2": user_id},
)

Two 32-bit ints. k1 = lock type; k2 = entity ID.

try_advisory_lock (non-blocking)

acquired = await session.scalar(
    text("SELECT pg_try_advisory_xact_lock(:key)"),
    {"key": 12345},
)
if not acquired:
    raise BusyError

Common patterns

Singleton job

async def run_singleton():
    async with AsyncSessionLocal() as session:
        async with session.begin():
            await session.execute(text("SELECT pg_advisory_xact_lock(42)"))
            await do_work()
# only one instance acquires at a time

Per-entity serialization

async def update_user(session, user_id, change):
    async with session.begin():
        await session.execute(
            text("SELECT pg_advisory_xact_lock(1, :uid)"),
            {"uid": user_id},
        )
        user = await session.get(User, user_id)
        apply_change(user, change)

Two updates to same user serialize; different users parallelize.

Visibility timeout (reaper)

# Hourly reset
await session.execute(
    update(Job)
    .where(Job.locked_at < datetime.utcnow() - timedelta(minutes=5))
    .values(locked_at=None, locked_by=None)
)

Recovers from worker crashes.

Lock timeout

async with session.begin():
    await session.execute(text("SET LOCAL lock_timeout = '5s'"))
    account = await session.scalar(select(Account).with_for_update())
    # raises after 5s if blocked

Like NOWAIT but with a wait period. SET LOCAL resets at commit.

Statement timeout (per-query cap)

async with session.begin():
    await session.execute(text("SET LOCAL statement_timeout = '60s'"))
    await session.execute(slow_query)

Deadlock retry

from sqlalchemy.exc import DBAPIError

async def with_retry(fn, attempts=3):
    for i in range(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 i == attempts - 1:
                raise
            await asyncio.sleep(0.1 * 2**i)

40P01 = deadlock detected.

Avoiding deadlocks

Acquire locks in consistent order:

def transfer(a, b, amt):
    first, second = sorted([a, b], key=lambda x: x.id)
    # lock first then second — same order across callers

NOWAIT vs SKIP LOCKED semantics

	Behavior
NOWAIT	Error if locked
SKIP LOCKED	Pretend the locked rows don’t exist
(default)	Wait until released

Pick by intent. For interactive UI: NOWAIT. For queue: SKIP LOCKED. For batch updates: default.

Common mistakes

Long-held lock during external HTTP call — blast radius.
No deadlock retry — 500 on contention.
Forgetting pg_advisory_unlock for session-scoped lock — leaks until close.
Mixing NOWAIT and SKIP LOCKED — different semantics.

SELECT FOR UPDATE#

NOWAIT#

SKIP LOCKED (queue pattern)#

FOR SHARE#

OF (lock specific tables in joins)#

Advisory locks (transaction-scoped)#

Advisory locks (session-scoped)#

Two-int advisory locks (namespaced)#

try_advisory_lock (non-blocking)#

Common patterns#

Singleton job#

Per-entity serialization#

Visibility timeout (reaper)#

Lock timeout#

Statement timeout (per-query cap)#

Deadlock retry#

Avoiding deadlocks#

NOWAIT vs SKIP LOCKED semantics#

Common mistakes#

Read this next#