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