Python Async With Statement

The async with statement is a control flow construct that evaluates and uses an asynchronous context manager, allowing the current coroutine to suspend execution and yield control during the context manager’s entry and exit phases.

Syntax

async with <expression> [as <target>] [, <expression> [as <target>] ...]:
    <block>

The async with statement can only be used within an asynchronous function (async def). The as <target> clause is optional. Multiple asynchronous context managers can be combined in a single async with statement by separating them with commas (e.g., async with a() as x, b() as y:). They are processed from left to right, which is semantically equivalent to nesting multiple async with statements.

The Asynchronous Context Manager Protocol

For an object to be compatible with async with, it must implement the Asynchronous Context Manager Protocol, which consists of two special methods:

__aenter__(self): An asynchronous method that sets up the context. The current coroutine executing the async with statement awaits this method before executing the inner block. If an as clause is provided, the awaitable’s result is bound to the <target> variable.
__aexit__(self, exc_type, exc_value, traceback): An asynchronous method that tears down the context. The current coroutine awaits this method after the inner block terminates, regardless of whether it exited normally or via an unhandled exception. This method is only called if __aenter__ completes successfully.

Both methods must return awaitable objects (typically by being defined as async def).

Class-Based Implementation

The mechanics of async with are explicitly defined by creating a class with the required dunder methods:

import asyncio

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

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

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Awaitable teardown phase
        await asyncio.sleep(0.1)
        self.state = "closed"
        # Returning False propagates exceptions; returning True suppresses them
        return False

async def main():
    async with AsyncResource() as resource:
        # The current coroutine resumes here after __aenter__ completes
        print(resource.state) 
    # The current coroutine resumes here after __aexit__ completes

if __name__ == "__main__":
    asyncio.run(main())

Generator-Based Implementation

Python provides the @asynccontextmanager decorator in the contextlib module to construct asynchronous context managers using asynchronous generator functions, bypassing the need for boilerplate class definitions.

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_resource_generator():
    # Code before the yield acts as __aenter__
    await asyncio.sleep(0.1)
    
    try:
        yield "resource_target"
    finally:
        # Code after the yield acts as __aexit__
        await asyncio.sleep(0.1)

async def main():
    async with async_resource_generator() as res:
        print(res)

if __name__ == "__main__":
    asyncio.run(main())

Execution Flow

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

Evaluates the <expression> to obtain the asynchronous context manager object.
Calls type(manager).__aenter__(manager).
Awaits the awaitable returned by __aenter__. If an exception is raised during this setup phase, the inner <block> is skipped, __aexit__ is not executed, and the exception propagates immediately.
Binds the awaited result to <target> (if the as keyword is used).
Executes the <block>.
Determines the exception state of the <block>. If an exception was raised during the block’s execution, exc_type, exc_val, and exc_tb are populated with the exception’s class, instance, and traceback, respectively. If the block executed successfully, all three variables are set to None.
Calls type(manager).__aexit__(manager, exc_type, exc_val, exc_tb).
Awaits the awaitable returned by __aexit__.
Evaluates the awaited result from __aexit__ if an exception was raised in the <block>. If the boolean value of the result evaluates to True, the exception is suppressed, and execution continues normally after the async with statement. If the result evaluates to False, the exception is propagated up the call stack.