Skip to content

Managing Dynamic Resource Sets with AsyncExitStack

A reporting job fans a query out to every database shard listed in configuration — three in staging, forty in production. Nested async with blocks cannot express "however many there are", so the code opens connections in a loop, appends them to a list, and closes them in a finally. It works until shard 27 is unreachable: the loop raises before reaching the try, and connections 1 through 26 stay open until the pool's idle timeout reaps them. Run the job every minute and the database hits its connection limit by lunchtime. contextlib.AsyncExitStack is the standard library's answer to this shape of problem. It lets you enter any number of async context managers imperatively, registers each one as soon as it is acquired, and unwinds whatever was acquired — in reverse order — no matter where the failure happened. This guide covers the patterns that go beyond the basic loop: plain cleanup callbacks, returning resources from a factory with pop_all(), application-lifetime containers, and what happens when cleanup itself fails.

Prerequisites

Shard c fails: what the stack releases 4 lanes over time. Shard c fails: what the stack releases shard a open held close shard b open held close shard c fails caller stack unwinds time → Only resources that opened are closed, newest first, before the exception propagates.

1. Acquire a runtime-sized set and unwind partial failures

Enter each resource through stack.enter_async_context(). The call awaits the manager's __aenter__ and, only if that succeeds, pushes its __aexit__ onto the stack. When the async with block exits — normally, by exception, or by cancellation — every registered exit runs in last-in, first-out order.

import asyncio
import contextlib

log: list[str] = []


class Shard:
    def __init__(self, name: str, fail: bool = False) -> None:
        self.name, self.fail = name, fail

    async def __aenter__(self) -> "Shard":
        await asyncio.sleep(0)                   # connect
        if self.fail:
            raise ConnectionError(f"{self.name} unreachable")
        log.append(f"open {self.name}")
        return self

    async def __aexit__(self, *exc) -> None:
        await asyncio.sleep(0)                   # disconnect
        log.append(f"close {self.name}")


async def query_all(names: list[str], bad: str | None = None) -> list[str]:
    async with contextlib.AsyncExitStack() as stack:
        shards = [await stack.enter_async_context(Shard(n, fail=n == bad)) for n in names]
        return [s.name for s in shards]


async def main() -> None:
    print(await query_all(["a", "b", "c"]), log)
    log.clear()
    try:
        await query_all(["a", "b", "c"], bad="c")
    except ConnectionError as exc:
        print("failed:", exc, log)


asyncio.run(main())

The successful run logs open a, open b, open c, close c, close b, close a. The failing run logs open a, open b, close b, close a: shard c never opened, so it is never closed, and the two that did open are released before the ConnectionError reaches the caller.

Verify: count open connections on the server before and after a run with an unreachable shard; the count returns to its baseline immediately rather than after an idle timeout.

2. Register cleanup for things that are not context managers

Many resources come with a close method rather than a context manager: a client with aclose(), a temporary directory to delete, a metrics flush, a lock to release, a task to cancel. push_async_callback() and callback() put arbitrary cleanup on the same stack, interleaved in acquisition order with the context managers.

import asyncio
import contextlib


async def run_batch(job_id: str) -> None:
    async with contextlib.AsyncExitStack() as stack:
        client = await make_client()                          # has aclose(), no __aexit__
        stack.push_async_callback(client.aclose)

        heartbeat = asyncio.create_task(send_heartbeats(job_id))
        stack.callback(heartbeat.cancel)                      # sync callback is fine too

        stack.push_async_callback(flush_metrics, job_id)      # arguments are bound now
        await process(client, job_id)

Register the cleanup on the line immediately after acquiring the resource. Any code placed between acquisition and registration is a window in which an exception — or a cancellation arriving at an await — leaks the resource, which is exactly the bug the stack exists to remove.

Verify: inject a failure inside process(); the logs show flush_metrics, then heartbeat cancellation, then client.aclose — the reverse of registration order.

One stack, mixed cleanup, reverse order 4 stacked layers from Registered last, runs first to Registered first, runs last. One stack, mixed cleanup, reverse order Registered last, runs first flush_metrics(job_id) push_async_callback Background task heartbeat.cancel callback Client client.aclose() push_async_callback Registered first, runs last Shard.__aexit__ enter_async_context Context managers and plain callbacks share one LIFO order.

3. Hand resources out of a factory with pop_all()

A function that opens resources for someone else to use has a problem: if it uses async with, everything closes when it returns; if it does not, a failure halfway through leaks. pop_all() solves both. Acquire inside a stack so failures unwind, then, once everything succeeded, move all the registered callbacks to a new stack and return it. The original stack exits empty and closes nothing.

import asyncio
import contextlib


async def open_shards(names: list[str]) -> tuple[list[Shard], contextlib.AsyncExitStack]:
    async with contextlib.AsyncExitStack() as stack:
        shards = [await stack.enter_async_context(Shard(n)) for n in names]
        return shards, stack.pop_all()           # success: transfer ownership to the caller


async def main() -> None:
    shards, owned = await open_shards(["x", "y"])
    async with owned:                             # caller now controls the lifetime
        print([s.name for s in shards])
    print(log)                                    # open x, open y, close y, close x


asyncio.run(main())

The returned stack is itself an async context manager, so the caller can use it in async with or call await owned.aclose() explicitly. If any __aenter__ in the factory raises, pop_all() is never reached, and the factory's own stack closes everything that had opened.

Who closes the resources? 2 columns contrasting async with stack, return stack.pop_all(). Who closes the resources? async with stack the function owns them closed when the block exits right for do-the-work code caller never sees them return stack.pop_all() the caller owns them closed when the caller exits right for factories failures still unwind Pop only after the last acquisition succeeded, never before.

Verify: the log shows both shards still open when open_shards returns and closed only when the caller's block exits; forcing the second shard to fail inside the factory closes the first one before the exception propagates.

4. Hold application-lifetime resources in one stack

Services open long-lived resources at startup — database pools, HTTP clients, a message consumer, background tasks — and must close them at shutdown in reverse order. A single AsyncExitStack owned by the application object is a clean home for them, and it plugs directly into lifespan hooks such as Starlette's or FastAPI's.

import asyncio
import contextlib
from dataclasses import dataclass, field


@dataclass
class AppResources:
    stack: contextlib.AsyncExitStack = field(default_factory=contextlib.AsyncExitStack)
    shards: list[Shard] = field(default_factory=list)

    async def start(self, shard_names: list[str]) -> None:
        try:
            for name in shard_names:
                self.shards.append(await self.stack.enter_async_context(Shard(name)))
            worker = asyncio.create_task(asyncio.sleep(3600), name="consumer")
            self.stack.push_async_callback(cancel_and_wait, worker)
        except BaseException:
            await self.stack.aclose()            # startup failed: release what opened
            raise

    async def stop(self) -> None:
        await self.stack.aclose()


async def cancel_and_wait(task: asyncio.Task) -> None:
    task.cancel()
    await asyncio.wait({task}, timeout=5)


@contextlib.asynccontextmanager
async def lifespan(app):                         # Starlette / FastAPI lifespan hook
    resources = AppResources()
    await resources.start(["primary", "replica"])
    app.state.resources = resources
    try:
        yield
    finally:
        await resources.stop()

The consumer task is registered after the pools, so it is stopped before them — it may still be using a connection while it finishes its current message. That ordering falls out of acquisition order automatically, which is why a single stack is safer than a hand-written shutdown function that has to be kept in sync with startup. The same ordering principle drives draining in-flight requests before shutdown.

Verify: start the app, then stop it; the consumer's cancellation is logged before either shard closes. Make the replica unreachable at startup and the primary is closed before the startup error surfaces.

5. Understand errors raised during teardown

Cleanup can fail: a socket is already broken, a flush times out. AsyncExitStack keeps unwinding after a failing exit, so the remaining resources are still released, and it preserves the chain of exceptions so the original error is not lost.

import asyncio
import contextlib


class BrokenOnClose(Shard):
    async def __aexit__(self, *exc) -> None:
        raise RuntimeError(f"{self.name}: close failed")


async def main() -> None:
    try:
        async with contextlib.AsyncExitStack() as stack:
            await stack.enter_async_context(Shard("ok"))
            await stack.enter_async_context(BrokenOnClose("bad"))
            raise ValueError("body failed")
    except Exception as exc:
        print(type(exc).__name__, exc)            # RuntimeError bad: close failed
        print("caused while handling:", repr(exc.__context__))   # ValueError('body failed')
        print(log)                                # ... close ok: later exits still ran


asyncio.run(main())

The exception that reaches the caller is the last one raised — here the RuntimeError from teardown — with the body's ValueError attached as __context__. Log handlers that print the full traceback show both; code that only inspects type(exc) sees the cleanup failure. If the cleanup error is less important than the original, catch and log it inside that resource's __aexit__ rather than letting it replace the real failure.

Verify: the output shows RuntimeError, a ValueError context, and close ok in the log — proving the healthy resource was released even though another resource's cleanup raised.

Verification

Dynamic resource management is correct when:

  • Partial acquisition never leaks: failing any single acquisition leaves server-side connection and file-descriptor counts at baseline.
  • Cleanup is registered immediately: no await sits between acquiring a resource and registering its cleanup.
  • Ownership is explicit: factories return a stack from pop_all(), and exactly one owner closes it.
  • Shutdown order mirrors startup: consumers and background tasks stop before the pools and clients they use.
  • Teardown errors are visible and contained: a failing cleanup is logged with its context, and every other resource is still released.

Pitfalls & edge cases

  • Calling pop_all() too early. Popping before every acquisition has succeeded disables the automatic unwind for the ones that follow. Pop as the last statement before returning.
  • Forgetting that enter_async_context needs an async manager. Passing a synchronous context manager raises TypeError; use stack.enter_context() for those, on the same stack.
  • Cancellation during teardown. If the task running aclose() is cancelled again, the remaining exits receive the cancellation at their first await. Shield critical flushes or give shutdown a bounded, uninterrupted window, as in using asyncio.shield to protect critical sections.
  • Suppressing exceptions by accident. An __aexit__ that returns a truthy value suppresses the exception for the whole stack, including the body's error. Return None unless suppression is intended.
  • One stack shared across tasks. Entering resources on the same stack from concurrent tasks makes the unwind order depend on scheduling. Give each task its own stack, or acquire sequentially at startup.

Frequently Asked Questions

When should I use AsyncExitStack instead of nested async with statements?

Use it when the number of resources is only known at runtime, when acquisition happens in a loop, or when some resources expose close methods instead of context managers. Nested async with blocks cannot express a variable count, while AsyncExitStack registers each resource as it is acquired and releases them all in reverse order.

Does AsyncExitStack clean up resources if one acquisition fails?

Yes. enter_async_context registers a resource's exit only after its aenter succeeds, so when a later acquisition raises, the stack unwinds and closes every resource that did open, in reverse order, before the exception reaches the caller. The failing resource itself is not closed because it never opened.

What does AsyncExitStack.pop_all() do?

It moves every registered cleanup callback onto a new AsyncExitStack and returns it, leaving the original stack empty. A factory can acquire resources inside its own stack for failure safety, then return pop_all() on success so the caller owns the resources and closes them with async with or aclose().

What happens if a cleanup callback raises inside AsyncExitStack?

The stack keeps running the remaining callbacks, so other resources are still released. The exception that finally propagates is the last one raised, with earlier exceptions, such as the original error from the block body, attached as its context, so full tracebacks still show the original failure.