Last updated: Apr 11, 2024
Reading time·3 min
The Python "RuntimeError: This event loop is already running" occurs when you try to nest the event loop.
To solve the error, install and use the nest_asyncio
package which allows
nested use of asyncio.run
and loop.run_until_complete
.
Here is an example of how the error occurs.
import asyncio async def my_coroutine(): print("Running my_coroutine") async def main(): loop = asyncio.get_event_loop() loop.run_until_complete(my_coroutine()) asyncio.run(my_coroutine()) # RuntimeError: This event loop is already running # sys:1: RuntimeWarning: coroutine 'my_coroutine' was never awaited asyncio.run(main())
The asyncio
module doesn't allow its event loop to be nested.
In an environment where the event loop is already running, you are not allowed to run tasks and wait for the result.
Trying to run a task in an environment where the event loop is already running causes the error.
The error commonly occurs in web servers, GUI applications and in Jupyter Notebooks.
nest_asyncio
module to solve the errorOne way to solve the error is to use the nest_asyncio module.
Open your terminal in your project's root directory and install the module.
pip install nest-asyncio # or with pip3 pip3 install nest-asyncio
Once the module is installed, import it and call the apply()
method to
initialize it.
import asyncio import nest_asyncio # 👇️ call apply() nest_asyncio.apply() async def my_coroutine(): print("Running my_coroutine") async def main(): loop = asyncio.get_event_loop() loop.run_until_complete(my_coroutine()) asyncio.run(my_coroutine()) asyncio.run(main())
The nest_asyncio
module patches asyncio
to allow nested use of the
asyncio.run()
and loop.run_until_complete()
methods.
The module patches the current event loop regardless of whether it is running or not.
Alternatively, you can solve the error by creating a single event loop outside
the main()
function.
import sys import asyncio if sys.version_info < (3, 10): loop = asyncio.get_event_loop() else: try: loop = asyncio.get_running_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) async def my_coroutine(): print("Running my_coroutine") async def main(): await my_coroutine() # 👇️ run the event loop outside the main() function loop.run_until_complete(main()) loop.close()
The if/else
statement assigns the running event loop to the loop
variable
depending on your Python version.
You can await your coroutines or tasks in the main()
function.
Notice that we called the loop.run_until_complete()
method outside the
main()
function.
The loop.run_until_complete method runs until the eventual result of the asynchronous operation has completed.
Make sure to call the method outside the main()
function to only create a
single event loop.
The loop.run_until_complete(main())
method call basically means "add main() to
be run by the event loop and run the event loop until main() is done".
The same approach can be used with the loop.create_task method.
import asyncio async def my_coroutine(index): print(f'ABC {index}') await asyncio.sleep(0.1) print(f'XYZ {index}') loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) async def main(): tasks = [] for index in range(3): tasks.append(asyncio.create_task(my_coroutine(index))) await asyncio.wait(tasks) loop.run_until_complete(main()) loop.close()
Running the code sample produces the following output.
ABC 0 ABC 1 ABC 2 XYZ 0 XYZ 1 XYZ 2
The loop.create_task
method takes a coroutine and schedules its execution.
The method returns a Task
object.
We append the tasks to a list and then await them using asyncio.wait()
.
Notice that we called the run_until_complete
method outside the main()
function.
You can learn more about the related topics by checking out the following tutorials: