Skip to content

Fix: Python asyncio Not Running / async Functions Not Executing

FixDevs ·

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 awaited

Or you get:

RuntimeError: no running event loop

Or:

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 traceback

Why 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().

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 — causes RuntimeError: 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() — using time.sleep() in async code blocks the event loop.

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 in if __name__ == "__main__"). Calling it multiple times or from inside an already-running coroutine raises errors. Everything else should use await.

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 content

Run 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 running

Fix — 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_asyncio
import nest_asyncio
nest_asyncio.apply()  # Patches asyncio to allow nested event loops

import asyncio
asyncio.run(fetch())  # Now works in Jupyter

Fix — 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 faster

Async-compatible alternatives to common blocking libraries:

BlockingAsync alternative
requestsaiohttp, httpx
time.sleep()asyncio.sleep()
open() / file I/Oaiofiles
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 of await asyncio.sleep(n) in async code. time.sleep() blocks the thread and the event loop — no other coroutine can run for n seconds. 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 running

Django (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 prints

Fixed — 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_job

Fix 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.py

Debug 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())

For Python async errors that show as RuntimeError: dictionary changed size during iteration, see Fix: Python RuntimeError: dictionary changed size during iteration. For general Python async connection errors, see Fix: Python asyncio RuntimeError: no running event loop.

F

FixDevs

Solo developer based in Japan. Every solution is cross-referenced with official documentation and tested before publishing.

Was this article helpful?

Related Articles