Skip to content

Writing Reentrant Async Context Managers

A reentrant context manager can be entered again while it is already active — a helper that opens a transaction calls another helper that opens one too, or a method holding a lock calls another method that takes the same lock. Most async context managers are not reentrant, and they fail in different ways when nested. Measured on Python 3.14: entering the same @asynccontextmanager object a second time, nested or after the first use, raised a confusing AttributeError: '_AsyncGeneratorContextManager' object has no attribute 'args'. Re-acquiring an asyncio.Lock in the task that holds it deadlocked, stopped only by a timeout. A reentrant lock that records its owning task let the owner nest and made a second task wait its turn. A "session" context manager that counted nesting depth on the instance opened one session for two concurrent tasks, merging independent units of work; keeping the depth in a ContextVar opened two, one per task, while still reusing it for nested calls within a task. With asyncpg, a nested conn.transaction() became a savepoint: an exception inside it rolled back only the inner insert, and the outer transaction committed ['outer', 'after']. This guide builds reentrancy deliberately.

Prerequisites

1. Recognise single-use context managers

An object returned by an @asynccontextmanager function wraps one generator, which can run once. Calling the function again creates a new manager; reusing the object does not work:

@asynccontextmanager
async def connection():
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

cm = connection()
async with cm:
    async with cm:            # same object, nested
        ...

Measured on Python 3.14: both nesting the object and re-entering it after it had exited raised AttributeError: '_AsyncGeneratorContextManager' object has no attribute 'args' — the error comes from the library's internals, not from a check meant for users, so it does not say "already used". The fix is to call the function each time (async with connection():), or, when the same resource should be shared by nested code, to design the manager for reentrancy as in the steps below. Storing a context manager object on self and entering it from several methods is the usual way to hit this.

Verify: no context manager object created by @asynccontextmanager is entered more than once.

2. Do not re-acquire asyncio.Lock in the same task

asyncio.Lock has no notion of an owner. A task that holds it and tries to acquire it again waits for itself:

lock = asyncio.Lock()

async def update_and_log():
    async with lock:
        await update()
        await log_change()            # also does `async with lock:` → waits forever

Measured: the nested acquire never completed, and asyncio.wait_for(..., 1) around it raised TimeoutError after a second. No error, no warning — the task is simply stuck, holding the lock everyone else is waiting for too. The best fix is structural: split each function into a public method that takes the lock and a private one that assumes it is held, so nested calls use the private version. Where call paths are too tangled for that, use a lock that knows its owner:

class RLock:
    """Reentrant lock: the owning task may acquire it again."""
    def __init__(self):
        self._lock = asyncio.Lock()
        self._owner: asyncio.Task | None = None
        self._depth = 0

    async def __aenter__(self):
        me = asyncio.current_task()
        if self._owner is me:
            self._depth += 1
            return self
        await self._lock.acquire()
        self._owner, self._depth = me, 1
        return self

    async def __aexit__(self, *exc):
        if self._owner is not asyncio.current_task():
            raise RuntimeError("released by a task that does not own it")
        self._depth -= 1
        if self._depth == 0:
            self._owner = None
            self._lock.release()

Measured with an owner that nested the lock and slept inside it, and a second task that tried to take it meanwhile: the order was owner nested, owner outer, then the other task — the owner re-entered freely and the other task waited until the outermost exit.

Verify: a function that nests the lock completes, and a concurrent task still waits for the outermost release.

What happens when the same manager is entered twice A grid of 6 rows by 3 columns. What happens when the same manager is entered twice manager nested in the same task used by two tasks @asynccontextmanager object AttributeError: no attribute 'args' - asyncio.Lock deadlock (TimeoutError after 1 s) serialised owner-tracking RLock allowed, depth counted other task waits session, depth on the instance reuses session 1 session for 2 tasks session, depth in a ContextVar reuses session 1 session per task asyncpg conn.transaction() savepoint - Python 3.14, asyncpg 0.31.0, Postgres 17.

3. Keep per-task nesting state in a ContextVar

A common reentrant pattern is a unit of work: the outermost async with opens a session or transaction, nested ones reuse it, and the outermost exit closes it. Counting depth on the manager instance seems to work in tests and breaks under concurrency, because the instance is shared by every task:

class SharedDepthSession:                       # wrong under concurrency
    async def __aenter__(self):
        if self.depth == 0:
            self.session = await open_session()
        self.depth += 1
        return self.session

Measured with two concurrent tasks each entering and nesting the same manager: one session was opened, and both tasks' nested calls received it — two independent units of work sharing one transaction, so a rollback in one would discard the other's writes. Keep the nesting state per task instead. A ContextVar is per task by construction, because each task runs in its own copy of the context:

_current = contextvars.ContextVar("unit_of_work", default=None)

class UnitOfWork:
    async def __aenter__(self):
        state = _current.get()
        if state is None:
            state = {"session": await open_session(), "depth": 0}
            _current.set(state)
        state["depth"] += 1
        return state["session"]

    async def __aexit__(self, exc_type, exc, tb):
        state = _current.get()
        state["depth"] -= 1
        if state["depth"] == 0:
            _current.set(None)
            await close_session(state["session"], commit=exc_type is None)

Measured: two sessions opened for the two tasks, and each task's nested call received that task's own session. Tasks created inside a unit of work inherit a copy of the context, and with it the outer task's session; decide deliberately whether child tasks should share it, as discussed in avoiding ContextVar leaks in background tasks.

Verify: concurrent tasks get separate sessions, and nested entries within one task get the same session.

4. Use savepoints for nested transactions

For databases, reentrancy usually means nested transactions, and the database already has the primitive: a savepoint. asyncpg's conn.transaction() uses one automatically when a transaction is already open on the connection:

async with conn.transaction():                          # BEGIN
    await conn.execute("INSERT INTO t VALUES ('outer')")
    try:
        async with conn.transaction():                  # SAVEPOINT
            await conn.execute("INSERT INTO t VALUES ('inner')")
            raise ValueError("inner step fails")         # ROLLBACK TO SAVEPOINT
    except ValueError:
        pass
    await conn.execute("INSERT INTO t VALUES ('after')")
                                                        # COMMIT

Measured against Postgres 17: the table held ['outer', 'after'] — the inner failure undid only the inner insert, and the outer transaction carried on and committed. This makes helper functions composable: each can open its own transaction block, and when called inside another one it becomes a recoverable step. The connection must be the same object for the nesting to be detected; acquiring a new connection from a pool inside the outer block starts an unrelated transaction, which is where a unit-of-work context like step 3's earns its keep. Cancellation inside a transaction block needs the care described in making database transactions cancellation-safe.

Verify: a failure inside a nested transaction block leaves the outer block's work intact.

Nested transaction blocks on one asyncpg connection A sequence of 6 messages between 4 participants. Nested transaction blocks on one asyncpg connection handler outer block inner block Postgres BEGIN; INSERT outer SAVEPOINT; INSERT inner ValueError ROLLBACK TO SAVEPOINT INSERT after; COMMIT rows: outer, after Reentrancy that the database implements for you.

5. Document which kind of reentrancy you provide

Reentrancy has several meanings, and callers need to know which one a manager offers. Make it explicit in the class and test it:

async def test_nested_entries_share_session():
    uow = UnitOfWork()
    async with uow as outer:
        async with uow as inner:
            assert inner is outer

async def test_tasks_get_separate_sessions():
    uow = UnitOfWork()
    async def unit():
        async with uow as s:
            await asyncio.sleep(0.01)
            return s
    a, b = await asyncio.gather(unit(), unit())
    assert a is not b

The three useful kinds are: single-use (a fresh manager per async with, the default for @asynccontextmanager); reentrant per task (an owner-tracking lock, a unit of work with ContextVar state); and nested with partial rollback (savepoints). A manager shared across tasks without per-task state is none of these, and its bugs only appear under concurrent load. Cleanup on exit must still follow the rules for __aexit__ — never swallowing cancellation — covered in handling exceptions in aexit.

Verify: tests cover nesting within a task and use from concurrent tasks for every reentrant manager.

What kind of reentrancy does this manager need? A decision on How is it nested with 4 outcomes. What kind of reentrancy does this manager need? How is it nested? never; one block each single-use, call it fresh reuse: AttributeError lock re-taken by its holder split methods, or owner-tracking RLock Lock deadlocks helpers sharing a session ContextVar depth 1 session per task nested database steps savepoints inner rollback only Per-task state is what makes reentrancy safe under concurrency.

Verification

Reentrant async context managers are correct when:

  • Single-use managers are created fresh for every async with.
  • Locks are either split into locked/unlocked methods or track their owning task.
  • Nesting state lives in a ContextVar, so concurrent tasks never share it by accident.
  • Nested database work uses savepoints, on the same connection.

Diagnostic Hook: log the session or transaction ID at the outermost enter together with the current task's name. Two task names logged against one session ID is the signature of shared nesting state — the bug that tests with a single task never show.

Pitfalls & edge cases

  • Re-entering an @asynccontextmanager object. Measured: an AttributeError that never mentions reuse.
  • Nesting asyncio.Lock. Measured: a silent deadlock.
  • Depth counters on a shared instance. Measured: two tasks in one session.
  • New pool connections inside a transaction. They are not nested; they are separate.

Frequently Asked Questions

Can I use the same async context manager object twice?

Not one created by @asynccontextmanager: re-entering it raised "AttributeError: '_AsyncGeneratorContextManager' object has no attribute 'args'" on Python 3.14. Call the function again for each async with.

Is asyncio.Lock reentrant?

No. Acquiring it again in the task that holds it deadlocked until a timeout. Split locked and unlocked methods, or use a lock that tracks its owning task.

How do I make a reentrant async context manager safe with multiple tasks?

Keep nesting depth and the shared resource in a ContextVar rather than on the instance: two concurrent tasks then got two sessions, while nested calls in one task reused its session.

Do nested asyncpg transactions work?

Yes: a nested conn.transaction() on the same connection is a savepoint. An exception inside it rolled back only the inner insert, and the outer transaction committed.