Fix: Python asyncio Not Running / async Functions Not Executing
Part of: Python Errors
Quick Answer
How to fix Python asyncio not running — coroutines never executing, RuntimeError no running event loop, mixing sync and async code, and common async/await mistakes in Python.
The Error
You write an async def function but calling it does nothing — the code never runs:
async def fetch_data():
print("Fetching...")
return "data"
fetch_data() # Nothing printed — the coroutine was never awaitedOr you get:
RuntimeError: no running event loopOr:
RuntimeError: This event loop is already running.Or:
RuntimeWarning: coroutine 'fetch_data' was never awaited
fetch_data()
RuntimeWarning: Enable tracemalloc to get the object allocation tracebackWhy This Happens
In Python, async def defines a coroutine function. Calling it returns a coroutine object — it does not execute the body of the function. To run the body, you must either await it inside another coroutine, or run it with an event loop using asyncio.run(). This is the root cause of “my async function never prints anything.” The interpreter happily creates a coroutine, and Python emits a runtime warning that gets buried in logs or never seen at all.
The second source of confusion is the event-loop model. Only one event loop runs in a given thread at a time. Calling asyncio.run() from inside a running loop fails because the loop cannot start a new loop inside itself. Jupyter notebooks, IPython, and async frameworks (FastAPI, Quart) all keep a loop running for you, so asyncio.run() is the wrong entry point in those contexts — you should await directly or use nest_asyncio.
The third trap is mixing blocking and non-blocking code. A single synchronous requests.get() call inside an async function stalls the entire event loop for the duration of the HTTP request. Other coroutines that were meant to run concurrently sit idle. The code “works” — it just runs sequentially instead of concurrently, defeating the purpose of asyncio. Fixing this requires either an async-native library or run_in_executor.
Common mistakes:
- Calling an async function without
await— you get a coroutine object, not the result. - Using
asyncio.run()inside an already-running event loop — causesRuntimeError: This event loop is already running. - Calling
asyncio.get_event_loop().run_until_complete()in contexts where a loop is already running (e.g., Jupyter notebooks). - Mixing synchronous and asynchronous code incorrectly — calling blocking code inside an async function blocks the event loop.
- Not awaiting
asyncio.sleep()— usingtime.sleep()in async code blocks the event loop.
Platform and Environment Differences
Python asyncio is portable, but the underlying event loop implementation, performance, and feature set vary by operating system and hosting environment. A subprocess example that works on macOS may raise NotImplementedError on Windows; a uvloop-accelerated benchmark on Linux may have no Windows equivalent.
Windows default event loop changed in Python 3.8. Starting in 3.8, Windows uses ProactorEventLoop by default (built on I/O Completion Ports), which supports subprocess and pipes that the older SelectorEventLoop did not. The reverse trade-off: ProactorEventLoop does not support some socket operations that work fine on SelectorEventLoop. If you get NotImplementedError on a UDP, raw socket, or specific selector API, fall back to the selector loop explicitly with asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()). Note that some Python libraries on Windows specifically require one or the other — read the library docs before forcing a policy.
uvloop on Linux and macOS only. uvloop is a high-performance asyncio replacement built on libuv. It is 2-4x faster than the default loop for typical I/O workloads, but it does not support Windows. If your code uses uvloop.install() unconditionally, Windows developers cannot run it. Guard the import with if sys.platform != "win32" and document the limitation. On macOS, uvloop works but you may need a recent build that matches your Python version.
Jupyter and IPython nested event loops. Jupyter runs its own loop in the kernel, so asyncio.run() always fails inside a notebook cell. Modern Jupyter (IPython 7+) supports top-level await, so just await my_coroutine() directly. If you must call asyncio.run() (for example, when porting a script unchanged), install nest_asyncio and call nest_asyncio.apply() — but note that nest_asyncio patches the loop globally, which can confuse libraries that detect the loop type. For broader Jupyter setup issues, see Fix: Jupyter not working.
AWS Lambda asyncio. Lambda’s Python runtime calls your handler synchronously. If your handler is async def, you must wrap it: def handler(event, context): return asyncio.run(async_handler(event, context)). Lambda reuses the execution context across invocations, so the event loop persists if you create it manually — using asyncio.run() fresh per invocation is safer but creates a new loop each call. For Lambda cold-start and timeout issues, see Fix: AWS Lambda timeout.
Gunicorn worker class for ASGI apps. Running FastAPI or Starlette with plain Gunicorn (the default sync worker) breaks asyncio entirely because the worker does not run an event loop. Use gunicorn -k uvicorn.workers.UvicornWorker app:app to get a proper async worker. With multiple workers, each has its own event loop and shared state must go through Redis, a database, or another external store — module-level globals are not shared. For setup issues with the worker itself, see Fix: Gunicorn not working.
Containers and signal handling. Inside Docker, the default asyncio signal handlers may not fire as expected because PID 1 has special signal semantics. Use tini or --init to ensure SIGTERM is delivered to your Python process, so graceful shutdown via asyncio.Event works during container stop.
Fix 1: Use asyncio.run() as the Entry Point
The correct way to run a top-level async function:
Broken — calling coroutine without awaiting:
import asyncio
async def main():
print("Hello from async")
await asyncio.sleep(1)
return "done"
result = main() # Returns coroutine object, prints nothing
print(result) # <coroutine object main at 0x...>Fixed — use asyncio.run():
import asyncio
async def main():
print("Hello from async")
await asyncio.sleep(1)
return "done"
result = asyncio.run(main())
print(result) # "done"asyncio.run() creates a new event loop, runs the coroutine until it completes, then closes the loop. It is the standard entry point for asyncio programs (Python 3.7+).
Pro Tip:
asyncio.run()should only be called once, at the top level of your program (typically inif __name__ == "__main__"). Calling it multiple times or from inside an already-running coroutine raises errors. Everything else should useawait.
Fix 2: Always await Coroutines
Every time you call an async def function, you must await the result to execute it:
Broken — missing await:
import asyncio
import aiohttp
async def fetch_url(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
# Missing await — result is a coroutine object, not the text
content = fetch_url("https://example.com")
print(content) # <coroutine object fetch_url at 0x...>
asyncio.run(main())Fixed — add await:
async def main():
content = await fetch_url("https://example.com")
print(content) # Actual HTML contentRun multiple coroutines concurrently with asyncio.gather():
async def main():
# Sequential — slow (waits for each one)
a = await fetch_url("https://example.com/a")
b = await fetch_url("https://example.com/b")
# Concurrent — fast (runs both at the same time)
a, b = await asyncio.gather(
fetch_url("https://example.com/a"),
fetch_url("https://example.com/b"),
)
print(a, b)asyncio.gather() runs coroutines concurrently. Use it whenever you have independent async operations that do not depend on each other’s results.
Fix 3: Fix “This event loop is already running” (Jupyter / IPython)
Jupyter notebooks run their own event loop. Calling asyncio.run() inside a Jupyter cell fails because a loop is already running:
# In a Jupyter notebook cell — raises RuntimeError
import asyncio
async def fetch():
await asyncio.sleep(1)
return "data"
asyncio.run(fetch()) # RuntimeError: This event loop is already runningFix — use await directly in Jupyter (Python 3.7+ in Jupyter):
# Jupyter supports top-level await in cells
result = await fetch()
print(result)Fix — use nest_asyncio for libraries that need asyncio.run():
pip install nest_asyncioimport nest_asyncio
nest_asyncio.apply() # Patches asyncio to allow nested event loops
import asyncio
asyncio.run(fetch()) # Now works in JupyterFix — use asyncio.get_event_loop().run_until_complete() (older approach):
loop = asyncio.get_event_loop()
result = loop.run_until_complete(fetch())Fix 4: Fix Blocking Code Inside Async Functions
A common mistake: calling blocking (synchronous) I/O inside an async function. This freezes the entire event loop until the blocking call completes — no other coroutines can run during that time:
Broken — blocking call in async function:
import asyncio
import time
import requests # Synchronous HTTP library
async def fetch_data(url: str):
response = requests.get(url) # Blocks the event loop for the entire request!
return response.text
async def main():
# These run sequentially despite being concurrent — requests.get blocks
results = await asyncio.gather(
fetch_data("https://example.com/a"),
fetch_data("https://example.com/b"),
)Fixed — use async-compatible libraries:
import asyncio
import aiohttp # Async HTTP library
async def fetch_data(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
results = await asyncio.gather(
fetch_data("https://example.com/a"),
fetch_data("https://example.com/b"),
)
# Both requests run concurrently — much fasterAsync-compatible alternatives to common blocking libraries:
| Blocking | Async alternative |
|---|---|
requests | aiohttp, httpx |
time.sleep() | asyncio.sleep() |
open() / file I/O | aiofiles |
psycopg2 (PostgreSQL) | asyncpg, psycopg3 |
pymysql (MySQL) | aiomysql |
redis-py (sync) | aioredis, redis.asyncio |
If you must run blocking code, use run_in_executor():
import asyncio
from concurrent.futures import ThreadPoolExecutor
import requests
executor = ThreadPoolExecutor(max_workers=10)
async def fetch_blocking(url: str):
loop = asyncio.get_event_loop()
# Run the blocking call in a thread pool — doesn't block the event loop
response = await loop.run_in_executor(executor, requests.get, url)
return response.text
async def main():
results = await asyncio.gather(
fetch_blocking("https://example.com/a"),
fetch_blocking("https://example.com/b"),
)Common Mistake: Using
time.sleep(n)instead ofawait asyncio.sleep(n)in async code.time.sleep()blocks the thread and the event loop — no other coroutine can run fornseconds.await asyncio.sleep(n)yields control back to the event loop, letting other coroutines run while waiting.
Fix 5: Fix asyncio with Frameworks (FastAPI, Django)
FastAPI is async-native — define route handlers as async def:
from fastapi import FastAPI
import asyncio
app = FastAPI()
@app.get("/data")
async def get_data():
# Can use await here
await asyncio.sleep(0.1) # Simulating async work
return {"data": "value"}
# Do NOT use asyncio.run() inside route handlers
# FastAPI's event loop is already runningDjango (with django.db sync ORM) — use sync_to_async for database calls:
from asgiref.sync import sync_to_async
from django.http import JsonResponse
from .models import User
async def user_view(request):
# Cannot call ORM directly in async view — wraps it in a thread
users = await sync_to_async(list)(User.objects.all())
return JsonResponse({"count": len(users)})Django 4.1+ supports async views natively with async def view functions, but the ORM is still synchronous — use sync_to_async or database_sync_to_async.
Fix 6: Fix asyncio Task Creation
Creating tasks with asyncio.create_task() schedules coroutines to run concurrently without waiting for each one:
Broken — task created but not awaited:
import asyncio
async def background_job():
await asyncio.sleep(5)
print("Done!")
async def main():
asyncio.create_task(background_job())
# main() returns immediately — background_job is cancelled before it finishes
print("Main done")
asyncio.run(main())
# Output: "Main done" — "Done!" never printsFixed — keep a reference and await:
async def main():
task = asyncio.create_task(background_job())
print("Main working...")
await task # Wait for the task to complete
print("Main done")Fixed — gather multiple tasks:
async def main():
tasks = [
asyncio.create_task(background_job()),
asyncio.create_task(another_job()),
]
print("Main working concurrently...")
await asyncio.gather(*tasks)
print("All done")For truly fire-and-forget tasks (not waiting for completion), store a reference to prevent garbage collection:
background_tasks = set()
async def main():
task = asyncio.create_task(background_job())
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)
# main continues without waiting for background_jobFix 7: Debug asyncio Issues
Enable asyncio debug mode to get detailed warnings about slow callbacks, unawaited coroutines, and other issues:
import asyncio
asyncio.run(main(), debug=True)Or via environment variable:
PYTHONASYNCIODEBUG=1 python your_script.pyDebug mode logs:
- Coroutines that took longer than 100ms (likely blocking calls).
- Tasks that were destroyed while still pending.
- Coroutines created but never awaited.
Use asyncio.current_task() and asyncio.all_tasks() to inspect running tasks:
async def debug_tasks():
tasks = asyncio.all_tasks()
for task in tasks:
print(f"Task: {task.get_name()}, done: {task.done()}")Catch unhandled task exceptions:
def handle_task_exception(loop, context):
exception = context.get("exception")
print(f"Unhandled exception in task: {exception}")
loop = asyncio.get_event_loop()
loop.set_exception_handler(handle_task_exception)Still Not Working?
Check Python version. asyncio.run() was added in Python 3.7. asyncio.TaskGroup (for structured concurrency) requires Python 3.11. Run python --version and upgrade if needed.
Check for synchronous generators or context managers. Using for item in async_generator (without async for) or with async_context_manager (without async with) silently produces wrong results. Use async for and async with for async iterators and context managers.
Check for event loop policy on Windows. Python 3.8+ on Windows uses ProactorEventLoop by default, which does not support some operations. If you get NotImplementedError on Windows for subprocess or UDP operations, set the policy explicitly:
import asyncio
import sys
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())Check for asyncio.run() called from a thread. asyncio.run() creates a new loop in the current thread. If you call it from a worker thread that already has a loop, or from a thread spawned by a framework that pre-populates a loop, the call fails or leaks loops. Use asyncio.run_coroutine_threadsafe(coro, loop) to schedule a coroutine onto a loop owned by another thread.
Check that __aiter__ and __anext__ are defined for async iterators. A class with __iter__ is a sync iterator; you need __aiter__ and __anext__ for async for. Mixing the two silently produces empty iteration or a TypeError deep in framework code.
Check that you are not awaiting a non-awaitable. Awaiting a regular function call (one that does not return a coroutine, Future, or Task) raises TypeError: object int can't be used in 'await' expression. Verify the function is async def or returns an awaitable; pure-sync functions wrapped with asyncio.to_thread() are awaitable, while plain returns are not.
For general Python async connection errors, see Fix: Python asyncio RuntimeError: no running event loop.
Solo developer based in Japan. Every solution is cross-referenced with official documentation and tested before publishing.
Was this article helpful?
Related Articles
Fix: aiosqlite Not Working — Single Writer, WAL Mode, Row Factory, and Connection Patterns
How to fix Python aiosqlite errors — database is locked, WAL mode for concurrent reads, foreign_keys PRAGMA, row factory for dict-like rows, connection per request vs pool, datetime detect_types, and FastAPI integration.
Fix: Tortoise ORM Not Working — Model Registration, Async Init, and Relationship Errors
How to fix Tortoise ORM errors — Tortoise.init not called, no module imported model, fetch_related missing, aerich migration setup, FastAPI integration patterns, and ConfigurationError missing connection.
Fix: asyncpg Not Working — Connection Pool, Prepared Statements, and Transaction Errors
How to fix asyncpg errors — connection refused localhost 5432, pool exhausted timeout, prepared statement does not exist, type codec not registered, JSON automatic conversion, and transaction rollback on exception.
Fix: Tenacity Not Working — Retries Not Firing, Exception Filters, and Async Support
How to fix Tenacity errors — retry decorator not retrying, stop_after_attempt vs stop_after_delay, retry_if_exception_type filter, async retry decorator, jitter for backoff, and RetryError unwrap original exception.