Last Updated on November 14, 2023
Context managers are a helpful construct in Python for automatically executing common code before and after a block.
The problem is that we cannot suspend and await coroutines at the entry and exit of a regular context manager.
The solution is to use an asynchronous context manager where the entry and exit code can be defined using coroutines and awaited automatically via the async with expression.
In this tutorial, you will discover how to develop and use asynchronous context managers.
After completing this tutorial, you will know:
- The difference between conventional and asynchronous context managers.
- How to develop asynchronous context managers for use in asyncio.
- How to use asynchronous context managers with the “async with” expression in asyncio.
Let’s get started.
What is an Asynchronous Context Manager
An asynchronous context manager is a Python object that implements the __aenter__() and __aexit__() methods.
Before we dive into the details of asynchronous context managers, let’s review classical context managers.
Context Manager
A context manager is a Python object that implements the __enter__() and __exit__() methods.
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
The __enter__() method defines what happens at the beginning of a block, such as opening or preparing resources, like a file, socket or thread pool.
The __exit__() method defines what happens when the block is exited, such as closing a prepared resource.
Typical uses of context managers include saving and restoring various kinds of global state, locking and unlocking resources, closing opened files, etc.
— With Statement Context Managers
A context manager is used via the “with” expression.
Typically the context manager object is created in the beginning of the “with” expression and the __enter__() method is called automatically. The body of the content makes use of the resource via the named context manager object, then the __aexit__() method is called automatically when the block is exited, normally or via an exception.
For example:
1 2 3 4 5 |
... # open a context manager with ContextManager() as manager: # ... # closed automatically |
This mirrors a try-finally expression.
For example:
1 2 3 4 5 6 7 8 |
... # create the object manager = ContextManager() try: manager.__enter__() # ... finally: manager.__exit__() |
Next, let’s take a look at asynchronous context managers.
Asynchronous Context Manager
Asynchronous context managers were introduced in “PEP 492 – Coroutines with async and await syntax“.
They provide a context manager that can be suspended when entering and exiting.
An asynchronous context manager is a context manager that is able to suspend execution in its __aenter__ and __aexit__ methods.
— Asynchronous Context Managers
The __aenter__ and __aexit__ methods are defined as coroutines and are awaited by the caller.
This is achieved using the “async with” expression.
You can learn more about the “async with” expression in the tutorial:
As such, asynchronous context managers can only be used within asyncio programs, such as within calling coroutines.
Next, let’s take a closer look at the difference between classical and asynchronous context managers.
Context Manager vs Asynchronous Context Manager
Both classical context managers and asynchronous context managers have a lot in common.
They both attempt to achieve the same effect of a try-finally expression with minimal code and a well-defined interface.
Classical context managers use the __enter__ and __exit__ methods, whereas asynchronous context managers use the __aenter__ and __aexit__ methods.
- Classical context managers: Implement __enter__() and __exit__() methods.
- Asynchronous context managers: Implement __aenter__() and __aexit__() methods.
Classical context managers may be used generally anywhere in Python programs, whereas asynchronous context managers may only be used within asyncio programs, such as within coroutines:
- Classical context managers: Used anywhere in python programs.
- Asynchronous context managers: Used in asyncio programs.
Classical context managers are used via the “with” expression, whereas asynchronous context managers are used via the “async with” expression.
- Classical context managers: Used via the “with” expression.
- Asynchronous context managers: Used via the “async with” expression.
Now that we know about asynchronous context managers, let’s look at how we might use them.
Run loops using all CPUs, download your FREE book to learn how.
How to Use Asynchronous Context Managers
In this section, we will explore how we can define, create, and use asynchronous context managers in our asyncio programs.
Define an Asynchronous Context Manager
We can define an asynchronous context manager as a Python object that implements the __aenter__() and __aexit__() methods.
Importantly, both methods must be defined as coroutines using the “async def” and therefore must return awaitables.
For example:
1 2 3 4 5 6 7 8 9 10 11 |
# define an asynchronous context manager class AsyncContextManager: # enter the async context manager async def __aenter__(self): # report a message print('>entering the context manager') # exit the async context manager async def __aexit__(self, exc_type, exc, tb): # report a message print('>exiting the context manager') |
Because each of the methods are coroutines, they may themselves await coroutines or tasks.
For example:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
# define an asynchronous context manager class AsyncContextManager: # 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) |
Use an Asynchronous Context Manager
An asynchronous context manager is used via the “async with” expression.
This will automatically await the enter and exit coroutines, suspending the calling coroutine as needed.
For example:
1 2 3 4 |
... # use an asynchronous context manager async with AsyncContextManager() as manager: # ... |
As such, the “async with” expression and asynchronous context managers more generally can only be used within asyncio programs, such as within coroutines.
You can learn more about the “async with” expression in the tutorial:
Now that we know how to use asynchronous context managers, let’s look at some worked examples.
Example of Using An Asynchronous Context Manager Manually
We can explore how to use an asynchronous context manager manually.
In this example, we will define an asynchronous context manager. The entry and exit methods of the manager will report a message and sleep for a moment. The messages will help us see when the methods are executed relative to our program code and then sleeps to simulate blocking I/O and show that these coroutines can be suspended when calling other coroutines.
The asynchronous context manager will be used manually.
That is, we will explicitly call the enter and exit methods, retrieve the awaitables from the coroutines, and await them directly.
This is a good example to show what exactly is happening automatically when we use more common usage patterns, such as the “async with” expression.
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
# SuperFastPython.com # example of using an asynchronous context manager manually import asyncio # define an asynchronous context manager class AsyncContextManager: # 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(): # create and use the asynchronous context manager manager = AsyncContextManager() # get the awaitable for entering the manager enter_awaitable = manager.__aenter__() # await the entry await enter_awaitable # execute the body print(f'within the manager') # get the awaitable for exiting the manager exit_awaitable = manager.__aexit__(None, None, None) # await the exit await exit_awaitable # start the asyncio program asyncio.run(custom_coroutine()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and creates an instance of our AsyncContextManager class.
Next, we call the __aenter__() coroutine to get an awaitable coroutine object.
The awaitable is then awaited, suspending the main() coroutine and executing the enter coroutine which reports a message and sleeps for a moment.
The main() coroutine resumes and reports a message.
It then calls the __aexit__() coroutine to get an awaitable coroutine object.
This second awaitable is awaited, suspending the main() coroutine, executing the coroutine that then reports a message and sleeps a moment.
The main() coroutine resumes and terminates the program.
This unrealistic example highlights what is happening automatically behind the scenes when we use a context manager with the “async with” expression, such as in the next section.
1 2 3 |
>entering the context manager within the manager >exiting the context manager |
Next, we will look at how to use a context manager via the “async with” expression.
Free Python Asyncio Course
Download your FREE Asyncio PDF cheat sheet and get BONUS access to my free 7-day crash course on the Asyncio API.
Discover how to use the Python asyncio module including how to define, create, and run new coroutines and how to use non-blocking I/O.
Example of an Asynchronous Context Manager and “async with”
We can explore how to use an asynchronous context manager via the “async with” expression.
In this example, we will update the above example to use the context manager in a normal manner.
We will use an “async with” expression and on one line, create and enter the context manager. This will automatically await the enter method.
We can then make use of the manager within the inner block. In this case, we will just report a message.
Exiting the inner block will automatically await the exit method of the context manager.
Contrasting this example with the previous example shows how much heavy lifting the “async with” expression does for us in an asyncio program.
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
# SuperFastPython.com # example of an asynchronous context manager via async with import asyncio # define an asynchronous context manager class AsyncContextManager: # 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(): # create and use the asynchronous context manager async with AsyncContextManager() as manager: # report the result print(f'within the manager') # start the asyncio program asyncio.run(custom_coroutine()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and creates an instance of our AsyncContextManager class in an “async with” expression.
This expression automatically calls the enter method and awaits the coroutine. A message is reported and the coroutine blocks for a moment.
The main() coroutine resumes and executes the body of the context manager, printing a message.
The block is exited and the exit method of the context manager is awaited automatically, reporting a message and sleeping a moment.
This highlights the normal usage pattern for an asynchronous context manager in an asyncio program.
1 2 3 |
>entering the context manager within the manager >exiting the context manager |
Next, we can look at what happens to an asynchronous context manager when an exception is raised.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of an Asynchronous Context Manager with an Exception
We can explore what happens to an asynchronous context manager when an exception is raised in the inner block.
In this example, we will update the above example that uses the asynchronous context manager via the “async with” expression, and raise an exception within the inner block.
The goal of this example is to show that regardless of how the inner block of the context manager is exited, the “async with” expression will ensure that the exit coroutine is awaited and executed.
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
# SuperFastPython.com # example of an asynchronous context manager that exits with an exception import asyncio # define an asynchronous context manager class AsyncContextManager: # 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(): # create and use the asynchronous context manager async with AsyncContextManager() as manager: # report the result print(f'within the manager') # fail with an exception raise Exception('Something bad happened') # start the asyncio program asyncio.run(custom_coroutine()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and creates an instance of our AsyncContextManager class in an “async with” expression.
This expression automatically calls the enter method and awaits the coroutine. A message is reported and the coroutine blocks for a moment.
The main() coroutine resumes and executes the body of the context manager, printing a message, and raising an exception.
The raised exception causes the context manager block to be exited. The exit method of the context manager is awaited automatically, reporting a message and sleeping a moment.
The exception then terminates the event loop of the asyncio program.
This highlights that the exit coroutine of an asynchronous context manager is awaited, regardless of how the inner block of the context manager is exited, such as with an unhandled exception.
1 2 3 4 5 6 |
>entering the context manager within the manager >exiting the context manager Traceback (most recent call last): ... Exception: Something bad happened |
Next, we can look at common errors when using asynchronous context managers.
Common Errors With Asynchronous Context Managers
It is common to stumble when getting started with asynchronous context managers.
In this section, we will explore a number of common errors when using asynchronous context managers and how they may be overcome.
Do you have an error not listed below?
Let me know in the comments below and I will do my best to help.
Example of an Asynchronous Context Manager and “with” Expression
We can explore what happens if we try to use an asynchronous context manager via the “with” expression.
In this example, we update the above example to use the asynchronous context manager normally, except with the wrong “with” expression, instead of the “async with” expression.
This is expected to result in an error as our AsyncContextManager does not implement the context manager interface, e.g. the __enter__() method, and will raise an exception.
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
# SuperFastPython.com # example of an asynchronous context manager via with (results in error) import asyncio # define an asynchronous context manager class AsyncContextManager: # 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(): # create and use the asynchronous context manager with AsyncContextManager() as manager: # report the result print(f'within the manager') # start the asyncio program asyncio.run(custom_coroutine()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and creates an instance of our AsyncContextManager class in a “with” expression.
This fails as the “with” expression expects to execute the __enter__() method on the object, which in this case does not exist. Instead, we have a __aenter__() method.
This results in an AttributeError that terminates the program.
The example highlights that asynchronous context managers cannot be used via the “with” expression and must be used via the “async with” expression within an asyncio program.
1 2 3 |
Traceback (most recent call last): ... AttributeError: __enter__ |
Next, we can look at trying to use an asynchronous context manager that does not return awaitables.
Example of an Asynchronous Context Manager without Awaitables
We can explore what happens if we try to use an asynchronous context manager that was defined using functions instead of coroutines.
In this example, we will use the asynchronous context manager normally, except update the definition of the class so that the enter and exit methods are Python functions, rather than coroutines.
This will cause an error as the “async with” expression will expect awaitables to be returned, rather than a return value, e.g. a None.
The complete example is listed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# SuperFastPython.com # example of an asynchronous context manager without awaitables (results in error) import asyncio # define an asynchronous context manager class AsyncContextManager: # enter the async context manager def __aenter__(self): # report a message print('>entering the context manager') # exit the async context manager def __aexit__(self, exc_type, exc, tb): # report a message print('>exiting the context manager') # define a simple coroutine async def custom_coroutine(): # create and use the asynchronous context manager async with AsyncContextManager() as manager: # report the result print(f'within the manager') # start the asyncio program asyncio.run(custom_coroutine()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and creates an instance of our AsyncContextManager class in an “async with” expression.
This fails as the “async with” expression executes to receive an awaitable when calling the __aenter__() coroutine, whereas the __aenter__() method was implemented as a normal function and returns a None value.
This highlights that the interface of an asynchronous context manager must be implemented using coroutines that return awaitables.
1 2 3 4 |
>entering the context manager Traceback (most recent call last): ... TypeError: 'async with' received an object from __aenter__ that does not implement __await__: NoneType |
Further Reading
This section provides additional resources that you may find helpful.
Python Asyncio Books
- Python Asyncio Mastery, Jason Brownlee (my book!)
- Python Asyncio Jump-Start, Jason Brownlee.
- Python Asyncio Interview Questions, Jason Brownlee.
- Asyncio Module API Cheat Sheet
I also recommend the following books:
- Python Concurrency with asyncio, Matthew Fowler, 2022.
- Using Asyncio in Python, Caleb Hattingh, 2020.
- asyncio Recipes, Mohamed Mustapha Tahrioui, 2019.
Guides
APIs
- asyncio — Asynchronous I/O
- Asyncio Coroutines and Tasks
- Asyncio Streams
- Asyncio Subprocesses
- Asyncio Queues
- Asyncio Synchronization Primitives
References
Takeaways
You now know how to create and use asynchronous context managers in Python.
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Do you have any questions?