Python Async Context Managers

Synchronous Python developers rely on with statements to manage resources. A file opened with with open(...) closes automatically when the block exits, even if an exception occurs. A lock acquired with with lock: releases automatically.

Async Python provides the same guarantee through async with and async context managers. The difference is that __aenter__ and __aexit__ are async methods that can contain await expressions, allowing setup and teardown to involve I/O without blocking the event loop. If you have read about context managers and async tasks, you understand both halves.

This article covers how they combine.

Async context managers are essential in concurrent code because resources shared across tasks must be cleaned up deterministically. An async database connection that is not returned to the pool, an async lock that is not released, or an HTTP session that is not closed will eventually exhaust resources and cause failures that are hard to diagnose, often long after the code that leaked the resource has already finished running.

async with: the syntax

async with works exactly like with except that it calls async methods and must appear inside an async def function:

python

import asyncio
 
class AsyncResource:
    async def __aenter__(self):
        print("Acquiring resource asynchronously...")
        await asyncio.sleep(0.1)
        print("Resource acquired.")
        return self
 
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Releasing resource asynchronously...")
        await asyncio.sleep(0.05)
        print("Resource released.")
        return False
 
    async def use(self):
        print("Using resource...")
        await asyncio.sleep(0.05)

With the class defined, using it looks just like a synchronous context manager except for the async keyword on def main() and the fact that setup and teardown can now await real I/O instead of blocking:

python

async def main():
    async with AsyncResource() as resource:
        await resource.use()
    print("Block exited.")
 
asyncio.run(main())

Both aenter and aexit print a message before and after their own asyncio.sleep() call, so the output shows the full acquire-use-release sequence in the order each step actually completes:

text

Acquiring resource asynchronously...
Resource acquired.
Using resource...
Releasing resource asynchronously...
Resource released.
Block exited.

aenter runs when the async with block is entered. Its return value is bound to the variable after as. aexit runs when the block exits, whether by normal completion, exception, or cancellation, which is exactly the same guarantee synchronous exit gives you inside an ordinary with block.

The three arguments to aexit are the exception type, value, and traceback, or all None if the block completed normally. Return True to suppress an exception, or False (the default) to let it propagate:

python

async def __aexit__(self, exc_type, exc_val, exc_tb):
    await self.cleanup()
    if exc_type is not None:
        print(f"Ignoring {exc_type.__name__}")
        return True
    return False

Practical example: async database connection

The most common real-world use of async context managers is managing database connections, where connecting and disconnecting both involve real network I/O that should not block the event loop while it happens:

python

import asyncio
class AsyncConnection:
    @classmethod
    async def connect(cls, dsn):
        await asyncio.sleep(0.1)
        conn = cls()
        conn.dsn = dsn
        conn.closed = False
        print(f"Connected to {dsn}")
        return conn
    async def close(self):
        if not self.closed:
            await asyncio.sleep(0.05)
            self.closed = True
            print("Connection closed.")
    async def execute(self, query):
        if self.closed:
            raise RuntimeError("Connection is closed")
        return f"Result of '{query}'"
    async def __aenter__(self):
        return self
    async def __aexit__(self, *args):
        await self.close()
        return False

Because aenter just returns the already-connected object and aexit calls close(), any code that already has a live AsyncConnection can drop it straight into an async with block without any extra setup:

python

async def main():
    conn = await AsyncConnection.connect("postgresql://localhost/mydb")
    async with conn:
        result = await conn.execute("SELECT 1")
        print(result)
 
asyncio.run(main())

The connection is always closed when the async with block exits. If execute() raises an exception, aexit still runs and closes the connection. This pattern prevents connection leaks in production code.

@asynccontextmanager: the generator shortcut

Writing a full class is verbose for simple resource management. contextlib.asynccontextmanager turns an async generator into an async context manager with a single yield. Everything before the yield is setup (aenter).

Everything after is teardown (aexit).

python

import asyncio
from contextlib import asynccontextmanager
 
@asynccontextmanager
async def managed_connection(dsn):
    print(f"Connecting to {dsn}...")
    await asyncio.sleep(0.1)
    conn = {"dsn": dsn, "closed": False}
 
    try:
        yield conn
    finally:
        print("Closing connection...")
        await asyncio.sleep(0.05)
        conn["closed"] = True
        print("Connection closed.")
 
async def main():
    async with managed_connection("postgresql://localhost/mydb") as conn:
        print(f"Using {conn['dsn']}")
 
asyncio.run(main())

The try/finally around yield ensures cleanup runs even if an exception occurs inside the async with block. This is the same pattern as synchronous @contextmanager.

A plain try/finally treats success and failure the same way, always running the same cleanup code regardless of what happened inside the block. Some resources need to react differently depending on whether the block succeeded or raised, and a database transaction is the clearest example: a successful block should commit, but a failing one should roll back instead.

python

@asynccontextmanager
async def transaction(conn):
    await conn.execute("BEGIN")
    try:
        yield conn
    except Exception:
        await conn.execute("ROLLBACK")
        raise
    else:
        await conn.execute("COMMIT")

If the block completes normally, the transaction commits. If any exception occurs, the transaction rolls back and the exception propagates. This pattern is ubiquitous in async database code.

Async context managers and cancellation

When a task is cancelled while inside an async with block, aexit still runs. This is critical because it means resources are released even during cancellation:

python

import asyncio
 
@asynccontextmanager
async def cancellable_resource():
    try:
        print("Setup...")
        await asyncio.sleep(0.1)
        yield "resource"
    finally:
        print("Teardown (always runs)...")
        await asyncio.sleep(0.05)

The task below gets cancelled one second before use_resource would naturally finish, which is exactly the scenario that tests whether finally really does run during a cancellation instead of only on normal exit:

python

async def main():
    task = asyncio.create_task(use_resource())
    await asyncio.sleep(0.05)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled")
 
async def use_resource():
    async with cancellable_resource() as r:
        await asyncio.sleep(1)
        print(f"Using {r}")
 
asyncio.run(main())

The teardown finally block runs even though the task was cancelled during the await asyncio.sleep(1). This is why async context managers are the correct way to manage async resources, not manual acquire and release calls: a manual approach would need its own try/except around every call site to catch CancelledError and release the resource, while the context manager guarantees it in one place.

Stacking async context managers

Real programs often need more than one resource open at the same time, such as a request that reads from one database and writes to another. Like with, you can stack multiple async with statements to enter several context managers together, and Python 3.10+ supports a parenthesized form so you do not have to nest a separate indented block for each one:

python

async def main():
    async with (
        managed_connection("db1") as conn1,
        managed_connection("db2") as conn2,
    ):
        result1 = await conn1.execute("SELECT 1")
        result2 = await conn2.execute("SELECT 2")
        print(result1, result2)

Both connections are acquired before entering the block and both are released when exiting. If acquiring conn2 fails, conn1 is released as part of aexit, so a failure partway through setup never leaves an earlier resource open.

Common async context managers in the standard library

Several asyncio primitives are async context managers:

• asyncio.TaskGroup: structured concurrency (Python 3.11+).
• asyncio.timeout(): set a timeout for a block (Python 3.11+).
• asyncio.Lock, asyncio.Semaphore, asyncio.Condition: async synchronization primitives.

Each of these already implements aenter and aexit for you, so you never need to write your own class just to use them safely inside an async with block. The example below shows all three in the same function to make the pattern obvious, though a real program would rarely combine them like this.

python

import asyncio
 
async def main():
    # Async lock as context manager
    lock = asyncio.Lock()
 
    async with lock:
        await do_protected_work()
 
    # Timeout context manager (Python 3.11+)
    async with asyncio.timeout(5):
        await slow_operation()
 
    # TaskGroup (Python 3.11+)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker("A", 0.1))
        tg.create_task(worker("B", 0.2))

The asyncio.timeout() context manager is particularly useful. It replaces the older asyncio.wait_for() pattern for blocks that contain multiple async operations, since wrapping several awaits in a single timeout block is clearer than wrapping each one individually:

python

async def fetch_with_timeout(url):
    async with asyncio.timeout(5):
        response = await http_client.get(url)
        data = await response.json()
        return data

If either get() or json() takes longer than 5 seconds combined, TimeoutError is raised and both are cancelled, since the timeout applies to the whole block rather than restarting for each individual await inside it.

Custom async context managers with state

When a resource has complex lifecycle, a class-based approach gives you more control than @asynccontextmanager can. A connection pool is a good example: it needs to track multiple live connections, limit how many exist at once, and expose helper methods beyond just entering and exiting the context, none of which fits cleanly into a single generator function with one yield.

python

import asyncio
 
class AsyncPool:
    def __init__(self, max_size=10):
        self.max_size = max_size
        self.sem = asyncio.Semaphore(max_size)
        self.connections = []
    async def acquire(self):
        await self.sem.acquire()
        if self.connections:
            return self.connections.pop()
        return await self._create_connection()
    async def release(self, conn):
        self.connections.append(conn)
        self.sem.release()
    async def _create_connection(self):
        await asyncio.sleep(0.1)
        return {"id": len(self.connections)}
    async def __aenter__(self):
        self.conn = await self.acquire()
        return self.conn
    async def __aexit__(self, *args):
        await self.release(self.conn)

The aenter and aexit methods are thin wrappers around acquire() and release(), which is what lets the same pool object be used directly in an async with statement instead of requiring callers to acquire and release connections manually:

python

async def main():
    pool = AsyncPool(max_size=3)
    async with pool as conn:
        print(f"Got connection: {conn}")
    print("Connection returned to pool.")
 
asyncio.run(main())

Each async with pool acquires a connection and releases it back to the pool on exit. The pool manages concurrency through a semaphore and reuses connections. The class encapsulates the pooling logic, and the async with syntax makes acquisition and release deterministic.