How Python Handles Async With and Context Managers

In Python, asynchronous context managers allow programs to allocate and release resources asynchronously without blocking the event loop during setup and teardown operations. Python implements this pattern using the async with statement, which relies on the asynchronous context management protocol defined by two coroutine methods: __aenter__() and __aexit__(). This article explains the internal mechanics of async with, the execution lifecycle of asynchronous context managers, and how to implement them effectively in modern Python applications.

The Asynchronous Context Management Protocol

Standard context managers in Python use synchronous __enter__() and __exit__() methods invoked by the standard with statement. When handling I/O-bound resources—such as network sockets, database connections, or file operations—blocking the thread while establishing or closing a connection degrades performance.

To solve this, Python defines the asynchronous context management protocol via PEP 492. An asynchronous context manager must implement two special methods:

  1. __aenter__(self): A coroutine that runs when entering the runtime context. The return value of this coroutine is bound to the target variable specified in the as clause of the async with statement.
  2. __aexit__(self, exc_type, exc_val, exc_tb): A coroutine that runs when exiting the runtime context. It receives information about any exception raised within the block. If it returns a truthy value, the exception is suppressed.

Execution Flow of async with

When Python encounters an async with statement, it translates the syntax into a deterministic series of coroutine calls:

async with EXPR as VAR:
    BLOCK

The underlying execution flow proceeds as follows:

  1. Expression Evaluation: Python evaluates EXPR to obtain the context manager object.
  2. Context Entry: Python invokes await EXPR.__aenter__().
  3. Variable Assignment: If the as VAR clause is present, the result returned by await EXPR.__aenter__() is assigned to VAR.
  4. Block Execution: The code inside BLOCK executes.
  5. Context Exit: Once the block finishes—whether normally or via an exception—Python invokes await EXPR.__aexit__(exc_type, exc_val, exc_tb).
  6. Exception Handling: If an exception occurs inside BLOCK, its type, value, and traceback are passed to __aexit__(). If __aexit__() evaluates to True, the exception is swallowed; otherwise, the exception re-raises after __aexit__() finishes.

Implementing an Asynchronous Context Manager

You can implement an asynchronous context manager either by defining a class with the required protocol methods or by using the contextlib module.

Class-Based Implementation

Creating a class with __aenter__ and __aexit__ provides complete control over the setup, execution, and cleanup phases:

import asyncio

class AsyncDatabaseSession:
    def __init__(self, host: str):
        self.host = host
        self.connection = None

    async def __aenter__(self):
        print(f"Connecting to {self.host}...")
        await asyncio.sleep(0.1)  # Simulate network latency
        self.connection = f"Connected({self.host})"
        return self.connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"Closing connection to {self.host}...")
        await asyncio.sleep(0.05)  # Simulate async teardown
        self.connection = None
        if exc_type is not None:
            print(f"Handled error: {exc_val}")
            return False  # Propagate exception

async def main():
    async with AsyncDatabaseSession("localhost:5432") as db:
        print(f"Executing query with {db}")

asyncio.run(main())

Generator-Based Implementation with contextlib

Python provides the @contextlib.asynccontextmanager decorator, which allows you to define an asynchronous context manager using an asynchronous generator function. This approach separates setup and teardown using a standard try...finally block around a yield expression:

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_resource(name: str):
    print(f"Acquiring {name}")
    await asyncio.sleep(0.1)
    resource = {"name": name, "status": "active"}
    try:
        yield resource
    finally:
        print(f"Releasing {name}")
        await asyncio.sleep(0.05)

async def main():
    async with managed_resource("Worker Pool") as res:
        print(f"Using {res['name']}")

asyncio.run(main())

Key Use Cases