Pattern from SQLAlchemy with a context manager

Lessons 02-03 covered the theory. Now the practice: implementing advisory locks correctly from Python with SQLAlchemy 2.0 async, wrapped in a context manager that guarantees automatic release (no repetitive manual try/finally) and correctly handles the "lock not available" case.

By the end you'll have a reusable app/services/advisory_locks.py module that any code in your app can import and use cleanly.

The pattern without a context manager (what you do NOT want)

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


async def cron_with_explicit_lock(session: AsyncSession):
    """Implementation without abstraction — verbose, error-prone."""

    # Acquire
    got = await session.scalar(
        text("SELECT pg_try_advisory_lock(:k)"),
        {"k": 1001}
    )
    if not got:
        return

    try:
        # Work
        await do_work(session)
    except Exception:
        # Re-raise but with cleanup
        raise
    finally:
        # ALWAYS release
        await session.execute(
            text("SELECT pg_advisory_unlock(:k)"),
            {"k": 1001}
        )

Verbose, repeated, easy to forget the unlock. Every cron, every job, every batch has this boilerplate.

The context manager — transaction-level version (recommended)

# app/services/advisory_locks.py
from contextlib import asynccontextmanager
from typing import AsyncIterator
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


@asynccontextmanager
async def advisory_xact_lock(
    session: AsyncSession,
    key: int,
) -> AsyncIterator[bool]:
    """Acquire a transaction-level advisory lock.

    Use within a transaction. The lock is automatically released when the
    transaction commits or rolls back.

    Yields:
        True if lock was acquired, False if someone else has it.

    Usage:
        async with session.begin():
            async with advisory_xact_lock(session, 1001) as got_lock:
                if not got_lock:
                    return
                # do critical work
            # lock released automatically on commit
    """
    got = await session.scalar(
        text("SELECT pg_try_advisory_xact_lock(:key)"),
        {"key": key}
    )
    yield bool(got)
    # No need to unlock — transaction commit/rollback handles it

Usage:

async def cron_refresh_mv(session: AsyncSession):
    async with session.begin():
        async with advisory_xact_lock(session, 1001) as got_lock:
            if not got_lock:
                print("Another instance running, skipping")
                return

            await session.execute(
                text("REFRESH MATERIALIZED VIEW CONCURRENTLY top_posts")
            )
        # Lock released on commit

Clean. The release is automatic.

Variant with a namespace (composite key)

@asynccontextmanager
async def advisory_xact_lock_ns(
    session: AsyncSession,
    namespace: int,
    resource_id: int,
) -> AsyncIterator[bool]:
    """Version with two ints (namespace + resource)."""
    got = await session.scalar(
        text("SELECT pg_try_advisory_xact_lock(:ns, :rid)"),
        {"ns": namespace, "rid": resource_id}
    )
    yield bool(got)


# Convenience: a namespace enum
class LockNamespace:
    CRON = 1
    JOB_QUEUE = 2
    REFRESH_MV = 3
    BATCH_IMPORT = 4


# Usage
async with advisory_xact_lock_ns(
    session, LockNamespace.CRON, hash("reindex_posts") & 0x7FFFFFFF
) as got:
    if got:
        await reindex()

Session-level variant with automatic cleanup

When you need session-level (a long-running worker), you can still wrap it:

@asynccontextmanager
async def advisory_session_lock(
    session: AsyncSession,
    key: int,
) -> AsyncIterator[bool]:
    """Session-level advisory lock with automatic cleanup.

    The lock is released when exiting the context.
    Use this for long-running operations spanning multiple transactions.

    WARNING: with PgBouncer transaction mode, session-level locks may
    persist between requests. Use advisory_xact_lock when possible.

    Usage:
        async with advisory_session_lock(session, 1001) as got_lock:
            if not got_lock:
                return
            # multiple transactions inside
            await tx_1(session)
            await tx_2(session)
        # lock released here
    """
    got = await session.scalar(
        text("SELECT pg_try_advisory_lock(:key)"),
        {"key": key}
    )

    if got:
        try:
            yield True
        finally:
            # ALWAYS release, even on exception
            await session.execute(
                text("SELECT pg_advisory_unlock(:key)"),
                {"key": key}
            )
    else:
        yield False

Usage:

async def long_running_worker(session: AsyncSession):
    async with advisory_session_lock(session, JOB_WORKER_KEY) as got_lock:
        if not got_lock:
            return

        # Multiple transactions OK
        for job in await fetch_jobs():
            async with session.begin():
                await process_job(session, job)
        # ... worker keeps running ...
    # Lock released

Variant that waits instead of skipping

Sometimes you want to wait for the lock instead of backing out:

@asynccontextmanager
async def advisory_xact_lock_blocking(
    session: AsyncSession,
    key: int,
    timeout_ms: int | None = None,
) -> AsyncIterator[None]:
    """Waits until it acquires the lock (no default TIMEOUT).

    If timeout_ms is set, it uses statement_timeout to abort the wait.
    """
    if timeout_ms:
        await session.execute(text(f"SET LOCAL statement_timeout = {timeout_ms}"))

    await session.execute(
        text("SELECT pg_advisory_xact_lock(:key)"),  # blocking version
        {"key": key}
    )

    yield  # No bool — we always get the lock (or it fails with a timeout)


# Usage
async with session.begin():
    try:
        async with advisory_xact_lock_blocking(session, 1001, timeout_ms=5000):
            await do_work()
    except OperationalError as e:
        if "statement timeout" in str(e):
            raise HTTPException(503, "Lock not available, try later")
        raise

Useful when the operation "always" needs to run but you want a timeout so you don't get blocked indefinitely.

Real-world case: a re-indexing cron with FastAPI

# app/tasks/reindex.py
import asyncio
from datetime import datetime, timezone
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import SessionLocal
from app.services.advisory_locks import advisory_xact_lock_ns, LockNamespace


REINDEX_LOCK_KEY = hash("cron:reindex_posts") & 0x7FFFFFFF


async def reindex_posts_cron():
    """Cron that recalculates tsvector on modified posts.

    Run via cron every 5 min. If previous run still active, skip.
    """
    async with SessionLocal() as session:
        async with session.begin():
            async with advisory_xact_lock_ns(
                session, LockNamespace.CRON, REINDEX_LOCK_KEY
            ) as got_lock:
                if not got_lock:
                    print(f"[{datetime.now(timezone.utc)}] Another reindex active, skip")
                    return

                print(f"[{datetime.now(timezone.utc)}] Starting reindex...")

                # The real work
                count = await reindex_modified_posts(session)

                print(f"[{datetime.now(timezone.utc)}] Reindexed {count} posts")
            # Lock released on commit


async def reindex_modified_posts(session: AsyncSession) -> int:
    """Find posts with a NULL search_vector and recalculate."""
    result = await session.execute(text("""
        UPDATE posts
        SET search_vector = to_tsvector('spanish', coalesce(title, '') || ' ' || coalesce(content, ''))
        WHERE search_vector IS NULL
        RETURNING id
    """))
    return result.rowcount


if __name__ == "__main__":
    asyncio.run(reindex_posts_cron())

Crontab:

*/5 * * * * cd /app && python -m app.tasks.reindex

If two crons overlap: the second gets got_lock = False and finishes without error. No Redis, no extra services.

Real-world case: an endpoint with idempotency

from fastapi import APIRouter, Depends, HTTPException, Header
from app.services.advisory_locks import advisory_xact_lock_ns, LockNamespace


router = APIRouter()


@router.post("/payments")
async def create_payment(
    data: PaymentData,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
    db: AsyncSession = Depends(get_db),
):
    """Endpoint protected against duplicate retries."""

    # Convert idempotency_key (string) to int for the advisory lock
    key_int = hash(idempotency_key) & 0x7FFFFFFF

    async with db.begin():
        async with advisory_xact_lock_ns(
            db, LockNamespace.JOB_QUEUE, key_int
        ) as got_lock:
            if not got_lock:
                # Another request with the same idempotency_key is in progress
                raise HTTPException(429, "Already processing this idempotency key")

            # Check if it was already processed (idempotency check)
            existing = await db.scalar(
                text("SELECT response FROM idempotency WHERE key = :k"),
                {"k": idempotency_key}
            )
            if existing:
                return existing

            # Process
            result = await charge_card(data)

            # Save for future retries
            await db.execute(
                text("INSERT INTO idempotency (key, response) VALUES (:k, :r)"),
                {"k": idempotency_key, "r": result.json()}
            )

            return result
        # Lock released on commit

Race between two requests with the same key: one processes, the other gets a 429. The client retries later and receives the cached response.

Testing the context manager

# tests/test_advisory_locks.py
import pytest
import asyncio
from app.services.advisory_locks import advisory_xact_lock


@pytest.mark.asyncio
async def test_lock_acquired_when_free(async_session):
    async with async_session.begin():
        async with advisory_xact_lock(async_session, 99999) as got:
            assert got is True


@pytest.mark.asyncio
async def test_lock_not_acquired_when_held(async_engine):
    """If another conn holds the lock, this one doesn't get it."""
    SessionLocal = async_sessionmaker(async_engine)

    # Conn 1 takes the lock
    async with SessionLocal() as session1:
        async with session1.begin():
            async with advisory_xact_lock(session1, 99999) as got1:
                assert got1 is True

                # Conn 2 tries — should fail
                async with SessionLocal() as session2:
                    async with session2.begin():
                        async with advisory_xact_lock(session2, 99999) as got2:
                            assert got2 is False


@pytest.mark.asyncio
async def test_lock_released_after_commit(async_engine):
    """After the commit, another conn can take it."""
    SessionLocal = async_sessionmaker(async_engine)

    # Conn 1 takes it and commits
    async with SessionLocal() as session1:
        async with session1.begin():
            async with advisory_xact_lock(session1, 99999) as got1:
                assert got1 is True

    # Conn 2 can now take it
    async with SessionLocal() as session2:
        async with session2.begin():
            async with advisory_xact_lock(session2, 99999) as got2:
                assert got2 is True

Run with pytest -v.

Traps and common mistakes

1. advisory_xact_lock outside a transaction.

# ❌ We're not inside session.begin()
async with advisory_xact_lock(session, 1001) as got:
    if got:
        await do_work()
# When is the lock released? The tx is implicit via SQLAlchemy
# but the behavior can surprise you

Always inside async with session.begin() or equivalent so it's clear when it's released.

2. Mixing session-level and transaction-level for the same key.

# Conn 1 takes session-level
await session.scalar(text("SELECT pg_advisory_lock(1)"))

# Conn 2 tries transaction-level with the SAME key
async with session2.begin():
    got = await session2.scalar(text("SELECT pg_try_advisory_xact_lock(1)"))
    # got = FALSE — the namespace is shared

Session-level and transaction-level share the same key namespace. Don't mix them.

3. A context manager that silences acquisition errors.

# If SELECT pg_try_advisory_xact_lock(...) fails due to a SQL bug,
# `got` can be None instead of a bool — be careful with `if got:`

Use bool(got) or got is True to check explicitly.

4. Not accounting for concurrency in tests.

Unit tests usually have a single conn. For a real mutual-exclusion test, you need multiple conns like in the example above.

5. Re-implementing the context manager in every place.

Define it once in app/services/advisory_locks.py and reuse it. Without centralization, each cron has its own version, with different bugs.

6. Forgetting the bool(got) cast.

session.scalar() can return None in edge cases (empty result). bool(got) prevents if got: from interpreting None as False, which is correct but confusing.

Exercise: implement and test the complete module

Step 1: create app/services/advisory_locks.py with the 3 variants:

# app/services/advisory_locks.py
from contextlib import asynccontextmanager
from typing import AsyncIterator
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


class LockNamespace:
    CRON = 1
    JOB_QUEUE = 2
    REFRESH_MV = 3
    BATCH_IMPORT = 4


@asynccontextmanager
async def advisory_xact_lock(session: AsyncSession, key: int) -> AsyncIterator[bool]:
    """Transaction-level advisory lock — autoreleased on commit."""
    got = await session.scalar(text("SELECT pg_try_advisory_xact_lock(:k)"), {"k": key})
    yield bool(got)


@asynccontextmanager
async def advisory_xact_lock_ns(
    session: AsyncSession, namespace: int, resource_id: int
) -> AsyncIterator[bool]:
    """Transaction-level with a namespace."""
    got = await session.scalar(
        text("SELECT pg_try_advisory_xact_lock(:n, :r)"),
        {"n": namespace, "r": resource_id}
    )
    yield bool(got)


@asynccontextmanager
async def advisory_session_lock(
    session: AsyncSession, key: int
) -> AsyncIterator[bool]:
    """Session-level with automatic cleanup."""
    got = await session.scalar(text("SELECT pg_try_advisory_lock(:k)"), {"k": key})
    if got:
        try:
            yield True
        finally:
            await session.execute(text("SELECT pg_advisory_unlock(:k)"), {"k": key})
    else:
        yield False

Step 2: tests with multiple concurrent sessions.

# tests/test_advisory_locks.py
# (code from the "Testing" section)

Run:

pytest tests/test_advisory_locks.py -v

Step 3: integrate into a real cron.

Take the app/tasks/reindex.py module (the example from the "Real-world case" section) and run the cron twice simultaneously:

# Terminal 1
python -m app.tasks.reindex &

# Terminal 2 (immediately)
python -m app.tasks.reindex

Expected output:

Terminal 1: [...] Starting reindex...
Terminal 2: [...] Another reindex active, skip

Step 4: simulate timeouts.

Implement the variant with a timeout:

@asynccontextmanager
async def advisory_xact_lock_blocking_timeout(
    session: AsyncSession, key: int, timeout_ms: int = 5000
) -> AsyncIterator[None]:
    await session.execute(text(f"SET LOCAL statement_timeout = {timeout_ms}"))
    await session.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": key})
    yield

Test: one conn takes the lock, another tries with a timeout, and after 5s gets an OperationalError.

# Timeout test
async def test_blocking_with_timeout():
    pool_a = await asyncpg.create_pool(...)
    async with pool_a.acquire() as conn_a:
        await conn_a.fetchval("SELECT pg_advisory_lock(99999)")  # session-level

        # Another conn tries blocking with a 1s timeout
        pool_b = await asyncpg.create_pool(...)
        async with pool_b.acquire() as conn_b:
            with pytest.raises(asyncpg.exceptions.QueryCanceledError):
                async with conn_b.transaction():
                    await conn_b.execute("SET LOCAL statement_timeout = 1000")
                    await conn_b.fetchval("SELECT pg_advisory_xact_lock(99999)")

        await conn_a.fetchval("SELECT pg_advisory_unlock(99999)")

Summary and next step

What you learned:

• A context manager eliminates acquisition/release boilerplate.
• advisory_xact_lock(session, key): transaction-level, auto-released on commit. Recommended.
• advisory_xact_lock_ns(session, namespace, resource_id): with a namespace to avoid collisions.
• advisory_session_lock(session, key): session-level with automatic try/finally.
• The blocking variant with a timeout: for cases where it "always has to run" but with a cap.
• Tests: they require multiple concurrent sessions.
• Real integration: cron, endpoint with idempotency.

Before moving on, you should be able to:

• Implement the app/services/advisory_locks.py module from scratch.
• Test it with simulated concurrency.
• Decide which variant to use based on the case.
• Integrate it into real endpoints/crons.

In the next lesson we get to the complete decision matrix: advisory locks vs Redis vs ZooKeeper/etcd. You'll learn concrete criteria to decide, cases where each one wins, and how to argue the decision on your team. It's the close of the advisory-locks block before moving on to savepoints.

Resources

1. PEP 343 — Context managers — theoretical foundation.
2. contextlib — Python docs — asynccontextmanager.
3. SQLAlchemy 2.0 — Async session lifecycle — reference.
4. pytest-asyncio fixtures — for tests with sessions.

Lesson 04 of 08 — Module 6 — Advanced PostgreSQL for Backend Guide