Async Context Manager

An asynchronous context manager is an object that defines the runtime context to be established when executing an async with statement. It allows for the asynchronous setup and teardown of resources by suspending execution and yielding control back to the event loop during the entry and exit phases.

To implement the asynchronous context management protocol, a Python class must define two special methods (dunder methods) that return awaitables:

• __aenter__(self): Asynchronously prepares the context and returns the resource to be bound to the as clause.
• __aexit__(self, exc_type, exc_val, exc_tb): Asynchronously cleans up the context. It receives exception details if an error occurred within the context block. Returning True suppresses the exception; returning False or None allows it to propagate.

Class-Based Implementation

The standard approach involves defining a class with async def for both protocol methods.

import asyncio

class AsyncResource:
    def __init__(self):
        self.state = "initialized"

    async def __aenter__(self):
        # Awaitable setup operations occur here
        await asyncio.sleep(0.1) 
        self.state = "active"
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Awaitable teardown operations occur here
        await asyncio.sleep(0.1)
        self.state = "closed"
        return False # Propagate exceptions

async def main():
    async with AsyncResource() as resource:
        # Context block execution
        pass

Execution Flow

When the Python interpreter encounters an async with statement, it executes the following sequence:

1. Evaluates the context expression to obtain the asynchronous context manager.
2. Awaits the __aenter__() method.
3. Binds the return value of __aenter__() to the target specified in the as clause (if provided).
4. Executes the internal block of the async with statement.
5. Awaits the __aexit__() method, passing in exception arguments if the block raised an error, or (None, None, None) if it completed successfully.

Generator-Based Implementation

Python provides a functional alternative via the contextlib.asynccontextmanager decorator. This allows you to define an asynchronous context manager using a single asynchronous generator function, avoiding boilerplate class definitions.

The generator must yield exactly once. Code before the yield acts as __aenter__, the yielded value is bound to the as target, and code after the yield (typically inside a finally block) acts as __aexit__.

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_resource_generator():
    # Setup phase (__aenter__)
    await asyncio.sleep(0.1)
    resource = {"status": "active"}
    
    try:
        # Suspend execution and yield control to the async with block
        yield resource
    finally:
        # Teardown phase (__aexit__)
        await asyncio.sleep(0.1)
        resource["status"] = "closed"

async def main():
    async with async_resource_generator() as resource:
        # Context block execution
        pass

In the generator pattern, exceptions raised inside the async with block are re-raised at the yield statement. Catching them within the generator allows you to suppress them, mirroring the behavior of returning True from __aexit__.

Master Python with Deep Grasping Methodology!

Learn More