Release Async Resources with contextlib

A function streams rows from a database, and it holds a pooled connection while it yields them. Your code needs only the first row, so it breaks out of the loop:

import asyncio
from collections.abc import AsyncGenerator, AsyncIterator
from contextlib import asynccontextmanager


@asynccontextmanager
async def acquire(name: str, log: list[str]) -> AsyncIterator[str]:
    log.append(f"acquire:{name}")  # stands in for taking a pooled connection
    try:
        yield name
    finally:
        log.append(f"release:{name}")


async def rows(log: list[str]) -> AsyncGenerator[int, None]:
    async with acquire("conn", log):
        for value in range(3):
            await asyncio.sleep(0)  # stands in for a database read
            yield value


async def main() -> None:
    log: list[str] = []
    async for value in rows(log):
        break  # you need only the first row
    print(log)


asyncio.run(main())

['acquire:conn']

After the loop, the connection is still checked out. The generator is paused at its yield, inside async with acquire(...), and the event loop closes it later, at a time you don't control. Under load, the pool runs out of connections. This article covers three contextlib tools that let you choose when cleanup runs.

Pair setup and cleanup with @asynccontextmanager

acquire uses @asynccontextmanager, which the previous article used to own a TaskGroup. It turns an async generator that yields once into a context manager for async with:

The code before yield runs when the block starts. It can await, for example to open a connection.
The value you yield is what as binds.
The finally runs when the block ends, whether the block finishes, raises or is cancelled.

Put the cleanup in finally. Without it, an error in the block skips the code after yield.

Close a stream when the block ends with aclosing

aclosing(stream) awaits stream.aclose() when its block ends:

from contextlib import aclosing


async def main() -> None:
    log: list[str] = []
    async with aclosing(rows(log)) as stream:
        async for value in stream:
            break
    print(log)


asyncio.run(main())

['acquire:conn', 'release:conn']

aclose() raises GeneratorExit at the paused yield. The error unwinds the generator, async with acquire(...) exits, and the release runs before your next line. A generator that never started has acquired nothing, so closing it skips its body.

Close a varying number of resources with AsyncExitStack

Sometimes you only know at runtime how many resources you need, such as one connection per shard. AsyncExitStack collects cleanups as you go and runs them in reverse order when its block ends. This example reuses acquire from the first one:

from contextlib import AsyncExitStack


async def flush(log: list[str]) -> None:
    await asyncio.sleep(0)  # stands in for writing a buffer
    log.append("flush")


async def copy_shards(names: list[str]) -> list[str]:
    log: list[str] = []
    async with AsyncExitStack() as stack:
        for name in names:
            await stack.enter_async_context(acquire(name, log))
        stack.push_async_callback(flush, log)
        log.append("copying")
    return log


print(asyncio.run(copy_shards(["a", "b"])))

['acquire:a', 'acquire:b', 'copying', 'flush', 'release:b', 'release:a']

await stack.enter_async_context(cm) enters an async context manager and returns what it yields. stack.enter_context(cm) does the same for a plain with one.
stack.push_async_callback(fn, *args) arranges for await fn(*args) to run at the end.
If a step fails partway, the stack still cleans up everything it opened so far.

Cleanup runs during cancellation too

When a task is cancelled, CancelledError passes through finally blocks and async with blocks like any other error, and their cleanup runs. Cleanup may await, but a second cancellation can interrupt it at that await, as the cancellation article showed. Keep cleanup short. In the exercise, the release doesn't await at all, so nothing can interrupt it.

Common mistakes

Writing stack.push_async_callback(flush(log)). That passes a coroutine, not a function. When the block ends, the stack tries to call it and raises TypeError: 'coroutine' object is not callable. Pass the function and its arguments separately.
Calling stream.aclose() without await. That only creates an awaitable, so the generator's finally does not run there. It runs later, at a time you don't control, often when asyncio.run shuts down. Python 3.13 and later print a RuntimeWarning about it, and 3.12 prints nothing.
Using contextlib.closing on an async generator. It calls close(), which an async generator doesn't have, so you get AttributeError. Use aclosing.

Check your understanding The exercise builds its report with "log": list(log), not "log": log. Why the copy? Show answer

The report should show the log as it was when you built it. Suppose your code left a stream open. When asyncio.run shuts down, it closes that stream, and its cleanup appends the release to the shared list after you built the report. A report that holds the list itself then shows the release and hides the bug. The copy keeps the state at that moment.

See the contextlib documentation for every helper in this article.