Patch asyncio to allow nested event loops
npx @tessl/cli install tessl/pypi-nest-asyncio@1.5.0A utility library that patches asyncio to allow nested event loops. This solves the common RuntimeError "This event loop is already running" that occurs when trying to run tasks in environments where an event loop is already active, such as web servers, GUI applications, and Jupyter notebooks.
pip install nest-asyncioimport nest_asyncioimport asyncio
import nest_asyncio
# Apply the patch to allow nested event loops
nest_asyncio.apply()
# Now you can use asyncio.run() and loop.run_until_complete()
# even when an event loop is already running
async def my_async_function():
await asyncio.sleep(0.1)
return "Hello, World!"
# This works even in environments with existing event loops
result = asyncio.run(my_async_function())
print(result)Patches asyncio to make event loops reentrant, allowing nested execution of asyncio.run() and loop.run_until_complete() calls.
def apply(loop=None):
"""
Patch asyncio to make its event loop reentrant.
Parameters:
- loop (optional): Specific event loop to patch. If None, patches the current event loop.
Returns:
None
Raises:
ValueError: If attempting to patch a non-asyncio.BaseEventLoop
"""import nest_asyncio
# Patch the current event loop
nest_asyncio.apply()import asyncio
import nest_asyncio
# Create and patch a specific loop
loop = asyncio.new_event_loop()
nest_asyncio.apply(loop)
asyncio.set_event_loop(loop)import nest_asyncio
import asyncio
# Apply patch to allow asyncio.run() in Jupyter cells
nest_asyncio.apply()
async def fetch_data():
await asyncio.sleep(1)
return "Data fetched"
# This works in Jupyter notebooks after applying the patch
result = asyncio.run(fetch_data())import nest_asyncio
import asyncio
# Apply patch for web server environments
nest_asyncio.apply()
async def background_task():
await asyncio.sleep(2)
return "Background task completed"
# Can now run tasks within web request handlers
# where an event loop is already running
def request_handler():
result = asyncio.run(background_task())
return resultimport asyncio
import nest_asyncio
nest_asyncio.apply()
async def outer_task():
print("Outer task started")
# This nested call works with the patch
result = asyncio.run(inner_task())
print(f"Inner task result: {result}")
return "Outer task completed"
async def inner_task():
await asyncio.sleep(0.1)
return "Inner task result"
# Run the nested structure
result = asyncio.run(outer_task())
print(result)The package validates input and may raise:
ValueError: "Can't patch loop of type <type>"This occurs when attempting to patch non-asyncio event loops.
If Tornado is imported before nest-asyncio, the package automatically makes Tornado aware of the pure-Python asyncio Future to ensure compatibility.
The patch preserves context variable behavior in nested executions, ensuring that context is properly maintained across nested async calls.
The patch is thread-aware and handles event loop thread identification correctly, preventing conflicts in multi-threaded applications.