Python asyncio ensure event loop utility function
Here’s a utility function that safely ensures an event loop is running in a Python asyncio application.
This is useful because of the discrepancy in expectations between the asyncio library and how it ends up being used in the real world in a lot of applications. The asyncio library reasonably expects users to be in control of the event loop for the whole application lifecycle, but this becomes complicated in various situations:
- Existing projects that were not built with asyncio in mind;
- Third party libraries that try to interact with the event loop;
- Different deployment environments requiring different application lifecycle structures;
- Different handling of the event loop between the application runtime and the
test runtime, for example with
pytest-asyncio. - A desire to be able to do an individual piece of work concurrently within a larger application that may or may not be using asyncio, for example providing a library function that does async work but needs to be callable from both async and synchronous applications.
Unfortunately it’s not safe or reliable to call functions such as
asyncio.get_event_loop() or asyncio.run_until_complete() without knowing
the current context.
The ensure_event_loop() below aims to be a safe and
reliable way to ensure an event loop is running and to access that event loop
in most contexts. It is important to test this thoroughly before deploying it
to production, though.
import asyncio
from asyncio import AbstractEventLoop
def ensure_event_loop() -> AbstractEventLoop:
"""
Ensure that an event loop is running, by taking an existing open
event loop, or creating and setting a new one.
This is helpful because no event loop is running by default, and
asyncio expects users to be in control of the top-level entry point
to the async program.
In reality that is often not feasible, due to various third party
libraries having their own interactions with the event loop which are
out of our control. In addition, control of the event loop is different
between tests (where pytest-asyncio manages the top-level entry point),
and when it runs in a deployment environment.
"""
if open_loop := __get_open_event_loop():
return open_loop
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
return new_loop
def _get_open_event_loop() -> AbstractEventLoop | None:
try:
loop = asyncio.get_event_loop()
if loop.is_running() and not loop.is_closed():
return loop
return None
except RuntimeError:
"""
Catching this RuntimeError is the only way to determine that there is
no event loop.
https://github.com/python/cpython/blob/main/Lib/asyncio/events.py#L708
"""
return None