asyncio.run() Cannot Be Called From A Running Event Loop Error

The error message “asyncio.run() cannot be called from a running event loop” typically occurs when we attempt to call asyncio.run() from within an already running event loop in Python. This error indicates that we are trying to nest or call asyncio.run() recursively, which is not allowed.

The asyncio.run() function is designed to be called once at the top level of your program to run the main coroutine.

Example to demonstrate the ‘asyncio.run() cannot be called from a running event loop’ error

Syntax:

import asyncio

async def my_coroutine():
    print("Running coroutine...")

async def main():
    loop = asyncio.get_event_loop()
    loop.create_task(my_coroutine())
    await asyncio.sleep(1)
    asyncio.run(main())  # This will cause the error

asyncio.run(main())

Scenarios for ‘asyncio.run() cannot be called from a running event loop’ error

The error message “asyncio.run() cannot be called from a running event loop” typically occurs when you try to call asyncio.run() from within an already running event loop. Let’s explore a couple of scenarios where this error may arise:

Nested Event Loops– In this scenario, you already have an event loop running when you attempt to call asyncio.run() within the main() coroutine. This causes the error because asyncio.run() is designed to create and run a new event loop. To resolve this issue, you should structure your code in a way that avoids calling asyncio.run() from within an already running event loop.

Syntax:

import asyncio

async def nested_coroutine():
    print("Running nested coroutine...")

async def main():
    loop = asyncio.get_event_loop()
    loop.create_task(nested_coroutine())
    await asyncio.sleep(1)
    asyncio.run(main())  # Error: Cannot call asyncio.run() from within a running event loop

asyncio.run(main())

Manual Event Loop Control– In this scenario, you manually control the event loop using asyncio.get_event_loop() and loop.run_until_complete(). However, within the main() coroutine, you mistakenly attempt to call loop.run_until_complete() again, resulting in the error. To resolve this issue, you should structure your code so that the main coroutine is only executed once using loop.run_until_complete() or asyncio.run().

Syntax:

import asyncio

async def nested_coroutine():
    print("Running nested coroutine...")

async def main():
    loop = asyncio.get_event_loop()
    loop.create_task(nested_coroutine())
    await asyncio.sleep(1)
    loop.run_until_complete(main())  # Error: Cannot call asyncio.run() from within a running event loop

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

To resolve the ‘asyncio.run() cannot be called from a running event loop’ error

To resolve this issue, one should refactor the code to avoid recursive or nested calls to asyncio.run(). If we need to run multiple coroutines concurrently or set up more complex event loop handling, we can manually create and control the event loop using asyncio.get_event_loop() and related methods.

Here’s an example of manually creating and running an event loop. It needs to be made sure to structure our code in a way that avoids calling asyncio.run() from within an already running event loop. In this example, await has been used instead of asyncio.run().

Syntax:

import asyncio

async def my_coroutine():
    print("Running coroutine...")

async def main():
    loop = asyncio.get_event_loop()
    loop.create_task(my_coroutine())
    await asyncio.sleep(1)
    loop.run_until_complete(main())

asyncio.run(main())

In the below exam, instead of using asyncio.run() directly, we are refactoring our code to use a different approach that doesn’t require calling asyncio.run(). We can manually create and manage the event loop using asyncio.get_event_loop() and loop.run_until_complete().

Syntax:

import asyncio

async def my_coroutine():
    # Your coroutine logic here

def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(my_coroutine())
    loop.close()

# Inside an existing event loop
main()  # Call `main()` instead of `asyncio.run()`

What is asyncio.run()?

The asyncio.run() function was introduced in Python 3.7 as part of the asyncio module. It was added to provide a convenient way to run a coroutine and manage the associated event loop. The purpose of asyncio.run() is to simplify the process of running an asyncio program by creating a new event loop, running a coroutine, and handling the cleanup of the event loop afterwards. It should be used as the entry point of an asyncio program.

Conclusion

To conclude, one should only call asyncio.run() once at the top level of your program to run the main coroutine. If we need to run multiple coroutines concurrently or set up more complex event loop handling, we have to use appropriate methods such as asyncio.gather() or asyncio.create_task().

By manually managing the event loop, you can control its lifecycle and avoid conflicts with an already running event loop. It should be called only from the main thread and not within an already running event loop.

References

1. Running an asyncio program

Follow us at PythonClear to learn more about solutions to general errors one may encounter while programming in Python.