Blocking calls in async def

The gotcha

Putting a synchronous blocking call (time.sleep, requests.get, open().read() on a slow disk) inside an async def function blocks the entire event loop. No other coroutine runs until it returns. The function looks async; it isn’t.

Minimal repro

import asyncio
import time

async def slow_task(name):
    print(f"{name} start")
    time.sleep(2)            # blocks the loop
    print(f"{name} done")

async def main():
    await asyncio.gather(
        slow_task("A"),
        slow_task("B"),
        slow_task("C"),
    )

asyncio.run(main())

You’d expect ~2 seconds total (three tasks running concurrently). You get ~6 seconds — they execute serially because time.sleep doesn’t yield to the loop.

Why it happens

asyncio is single-threaded cooperative concurrency. Each coroutine runs until it hits an await, at which point it yields control to the loop. time.sleep is a C function that suspends the OS thread — there’s no yield point for the loop to schedule something else.

The same applies to:

• requests.get(...) (use httpx.AsyncClient or aiohttp)
• open(...).read() on slow filesystems (use aiofiles)
• psycopg2.execute(...) (use asyncpg or async SQLAlchemy)
• redis.Redis().get(...) (use redis.asyncio)
• CPU-bound work (use a process pool)

How to fix

1. Use the async version of the library

async def slow_task(name):
    await asyncio.sleep(2)   # yields to the loop

import httpx
async with httpx.AsyncClient() as client:
    # instead of requests.get
    r = await client.get(url)

This is always the first thing to try.

2. Run blocking code in a thread

When no async version exists (3rd-party C library, legacy code):

import asyncio

def blocking_lib_call(x):
    # synchronous, can't be made async
    return some_lib.process(x)

async def main():
    result = await asyncio.to_thread(blocking_lib_call, 42)

asyncio.to_thread (3.9+) runs the function in a thread pool and yields the coroutine that resolves to its return value. The event loop keeps running other tasks while the thread blocks.

Older Python:

loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, blocking_lib_call, 42)

3. CPU-bound work — use a process pool

Threads don’t help for CPU-bound work due to the GIL. Use processes:

from concurrent.futures import ProcessPoolExecutor

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_fn, data)

How to detect blocking calls

asyncio.run(main(), debug=True)

Or set the env var:

PYTHONASYNCIODEBUG=1 python app.py

Debug mode logs warnings when callbacks take longer than slow_callback_duration (default 0.1s), and when coroutines are never awaited. In production, aiomonitor or APM tools (Datadog, Sentry) flag long-blocking event-loop callbacks.

Common variants

time.sleep masquerading as asyncio.sleep:

import time as asyncio   # someone's terrible alias
await asyncio.sleep(1)   # this is time.sleep — TypeError actually, but if it weren't...

Mixing sync ORM with async framework:

@app.get("/users")
async def get_users():
    # SQLAlchemy sync → blocks!
    return User.query.all()

This is the #1 FastAPI / async-Django performance bug. Either go fully async (async with AsyncSession() as s: await s.execute(...)) or use a sync route handler — but don’t mix.

Logging file handlers:

logging.FileHandler writes synchronously. Under high throughput on slow disks, this blocks the loop. Use QueueHandler + QueueListener to offload, or aiologger.