How to Use the "async with" Expression in Python

November 5, 2022 Python Asyncio

You can use the async with expression to use asynchronous context managers in coroutines in asyncio programs.

In this tutorial, you will discover the asyncio async with expressions for asynchronous context managers in Python.

Let's get started.

What is async with

The "async with" expression is for creating and using asynchronous context managers.

An asynchronous context manager is a context manager that is able to suspend execution in its enter and exit methods.
-- The async with statement

It is an extension of the "with" expression for use in coroutines within asyncio programs.

The "async with" expression is just like the "with" expression used for context managers, except it allows asynchronous context managers to be used within coroutines.

In order to better understand "async with", let's take a closer look at asynchronous context managers.

What is an Asynchronous Context Manager

An asynchronous context manager is a context manager that can await the enter and exit methods.

Before we dive into asynchronous context managers, let's review context managers.

Context Manager

Recall that a context manager uses an object that implements the __enter__() and __exit__() methods via the "with" expression.

A context manager is an object that defines the runtime context to be established when executing a with statement. The context manager handles the entry into, and the exit from, the desired runtime context for the execution of the block of code.
-- With Statement Context Managers

Context managers are useful when using resources as they can be created and initialized at the beginning of the block, used within the block, and closed automatically when exiting the block.

Importantly, the resource can be closed regardless of how the block is exited, such as normally, with a return statement, or with a raised error or exception.

Common examples for context managers include files, sockets, and thread pools.

For example, we might create and use a thread pool via the context manager interface.

...
# create and use the thread pool
with ThreadPool() as pool:
	# use the thread pool...
# closed automatically

This is equivalent to something like:

...
# create the thread pool
pool = ThreadPool()
try:
	# use the thread pool...
finally:
	# close the pool
	pool.close()

Next, let's take a closer look at asynchronous context managers.

Asynchronous Context Manager

An asynchronous context manager must implement the __aenter__() and __aexit__() methods.

It must also only be used within a coroutine.

An asynchronous context manager is a context manager that is able to suspend execution in its enter and exit methods.
-- Asynchronous Context Managers and "async with"

Unlike a traditional context manager, the asynchronous context manager may await when entering and exiting the context manager's block.

This allows the caller to suspend execution and schedule and wait upon coroutines to perform tasks, such as preparing resources required within the block.

This might include opening a socket or preparing a file for use.

An asynchronous context manager must be used via the "async with" expression.

How to use "async with"

The async with expression allows a coroutine to create and use an asynchronous version of a context manager.

For example:

...
# create and use an asynchronous context manager
async with AsyncContextManager() as manager:
	# ...

This is equivalent to something like:

...
# create or enter the async context manager
manager = await AsyncContextManager()
try:
	# ...
finally:
	# close or exit the context manager
	await manager.close()

Notice that we are implementing much the same pattern as a traditional context manager, except that creating and closing the context manager involve awaiting coroutines.

This suspends the execution of the current coroutine, schedules a new coroutine and waits for it to complete.

As such an asynchronous context manager must implement the __aenter__() and __aexit__() methods that must be defined via the async def expression. This makes them coroutines themselves which may also await.

async with vs with

The "async with" and "with" expressions are very similar.

The "with" expression is for using traditional context managers that implement the __enter__() and __exit__() methods.

The "async with" expression is for use with asynchronous context managers that implement the __aenter__() and __aexit__() methods.

To make this concrete, let's look at some worked example

Example of async with and an Asynchronous Context Manager

We can explore how to use the "async with" expression with a custom asynchronous context manager.

The example below defines a custom asynchronous context manager that reports a message and sleeps in the enter and exit methods. A coroutine is defined that creating and using the asynchronous context manager via the async with expression.

# SuperFastPython.com
# example of async with and an asynchronous context manager
import asyncio

# define an asynchronous context manager
class CustomContextManager:
    # enter the async context manager
    async def __aenter__(self):
        # report a message
        print('>entering the context manager')
        # block for a moment
        await asyncio.sleep(0.5)

    # exit the async context manager
    async def __aexit__(self, exc_type, exc, tb):
        # report a message
        print('>exiting the context manager')
        # block for a moment
        await asyncio.sleep(0.5)

# define a simple coroutine
async def custom_coroutine():
    # report a message
    print('before the context manager')
    # create and use the asynchronous context manager
    async with CustomContextManager() as manager:
        # report the result
        print(f'within the manager')
    # report a message
    print('after the context manager')

# start
asyncio.run(custom_coroutine())

Running the example first creates the custom coroutine and uses it as the entry point into the asyncio program.

The custom coroutine then uses an async with expression to create and use the CustomContextManager class.

The coroutine first awaits the __aenter__() method, which reports a message and blocks for a moment.

The body of the context manager block then executes, reporting a message.

Finally, the coroutine awaits the __aexit__() method, which also reports a message and blocks.

before the context manager
>entering the context manager
within the manager
>exiting the context manager
after the context manager

Example of async with and a Traditional Context Manager

A coroutine can use a traditional context manager if needed, but the context manager must be used via the "with" expression, not the "async with" expression. Otherwise, an error will be raised.

We can demonstrate this by updating the CustomContextManager to implement the __enter__() and __exit__() methods for a traditional context manager and try to use it via the "async with" expression.

The complete example is listed below.

# SuperFastPython.com
# example of async with and a traditional context manager
import asyncio

# define a traditional context manager
class CustomContextManager:
    # enter the context manager
    def __enter__(self):
        # report a message
        print('>entering the context manager')

    # exit the context manager
    def __exit__(self, exc_type, exc, tb):
        # report a message
        print('>exiting the context manager')

# define a simple coroutine
async def custom_coroutine():
    # report a message
    print('before the context manager')
    # create and use the context manager
    async with CustomContextManager() as manager:
        # report the result
        print(f'within the manager')
    # report a message
    print('after the context manager')

# start
asyncio.run(custom_coroutine())

Running the example results in an error as expected.

This is because the "async with" expression expects to operate upon an asynchronous context manager via the __aenter__() and __aexit__() methods that are themselves coroutines.

These methods were not present and an error was raised.

This highlights that the "async with" expression must be used with asynchronous context managers only, not with traditional context managers.

before the context manager
Traceback (most recent call last):
  ...
AttributeError: __aenter__

Example of with and an Asynchronous Context Manager

Similarly, we cannot use an asynchronous context manager via the "with" expression.

It must be used via the "async with" expression.

We can demonstrate this by updating the first example above to attempt to use the asynchronous context manager via the "with" expression in the coroutine, which results in an error.

The complete example is listed below.

# SuperFastPython.com
# example of with and an asynchronous context manager
import asyncio

# define an asynchronous context manager
class CustomContextManager:
    # enter the async context manager
    async def __aenter__(self):
        # report a message
        print('>entering the context manager')
        # block for a moment
        await asyncio.sleep(0.5)

    # exit the async context manager
    async def __aexit__(self, exc_type, exc, tb):
        # report a message
        print('>exiting the context manager')
        # block for a moment
        await asyncio.sleep(0.5)

# define a simple coroutine
async def custom_coroutine():
    # report a message
    print('before the context manager')
    # create and use the asynchronous context manager
    with CustomContextManager() as manager:
        # report the result
        print(f'within the manager')
    # report a message
    print('after the context manager')

# start
asyncio.run(custom_coroutine())

Running the example results in an error as expected.

In this case, the "with" expression expects a traditional context manager object with the __enter__() and __exit__() methods that were not present in this case.

This highlights that an asynchronous context manager must be used via the "async with" expression only; they cannot be used via the "with" expression.

before the context manager
Traceback (most recent call last):
  ...
AttributeError: __enter__

Takeaways

You now know about the async with expression for asynchronous context managers.

If you enjoyed this tutorial, you will love my book: Python Asyncio Jump-Start. It covers everything you need to master the topic with hands-on examples and clear explanations.