FastAPI: async def vs def: the concurrency model

FastAPI runs async def endpoints on the event loop, one thread juggling every request, and plain def endpoints in a thread pool. Both can be fast. The disaster is a blocking call inside async def: time.sleep, requests.get, a synchronous database driver, a heavy loop. It freezes the loop, and every request in that process waits, including health checks. This part measures exactly that, so you never have to learn it in an incident.

1. Where your code runs

2. Measure it

Six endpoints that “do 0.2 s of waiting” or some CPU work in different ways. Each gets a burst of concurrent requests while a /health probe measures how long an unrelated, trivial request waits. The client talks to the app in-process over ASGI, so the numbers show the app’s own behaviour:

the same wait, four ways, and what it does to everyone else

import asyncio
import time

import anyio.to_thread
import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/async-await")
async def async_await():
    await asyncio.sleep(0.2)             # non-blocking wait: the loop serves others meanwhile
    return "ok"

@app.get("/async-blocking")
async def async_blocking():
    time.sleep(0.2)                      # BLOCKING call inside async def: the loop is frozen
    return "ok"

@app.get("/sync-blocking")
def sync_blocking():
    time.sleep(0.2)                      # blocking, but in a def endpoint: runs in the thread pool
    return "ok"

@app.get("/cpu-async")
async def cpu_async():
    return sum(i * i for i in range(2_000_000))      # pure-Python CPU work on the loop

@app.get("/cpu-sync")
def cpu_sync():
    return sum(i * i for i in range(2_000_000))      # the same work in a thread

@app.get("/health")
async def health():
    return "ok"

async def burst(client, path, n):
    start = time.perf_counter()
    await asyncio.gather(*(client.get(path) for _ in range(n)))
    return time.perf_counter() - start

async def health_latency_during(client, path, n):
    async def probe():
        start = time.perf_counter()
        await asyncio.sleep(0.05)                      # fire the probe while the burst is running
        await client.get("/health")
        return time.perf_counter() - start - 0.05
    results = await asyncio.gather(probe(), *(client.get(path) for _ in range(n)))
    return results[0]

async def main():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test", timeout=60) as client:
        await client.get("/health")
        print(f"{'endpoint':16}{'20 concurrent':>15}{'/health waits':>16}")
        for path in ("/async-await", "/async-blocking", "/sync-blocking"):
            total = await burst(client, path, 20)
            waited = await health_latency_during(client, path, 10)
            print(f"{path:16}{total:13.2f} s{waited * 1000:13.0f} ms")
        for path in ("/cpu-async", "/cpu-sync"):
            waited = await health_latency_during(client, path, 4)
            print(f"{path:16}{'':15}{waited * 1000:13.0f} ms")
        limit = anyio.to_thread.current_default_thread_limiter().total_tokens
        print(f"\nthread pool size: {limit};  60 concurrent /sync-blocking: {await burst(client, '/sync-blocking', 60):.2f} s")

asyncio.run(main())

Output

endpoint          20 concurrent   /health waits
/async-await             0.21 s            9 ms
/async-blocking          4.03 s         1964 ms
/sync-blocking           0.24 s           15 ms
/cpu-async                              1179 ms
/cpu-sync                                 92 ms

thread pool size: 40;  60 concurrent /sync-blocking: 0.44 s

3. The decision rules

calling a blocking library from async code, the right way

import asyncio
import time

import httpx
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool

def legacy_pdf_render(order_id: int) -> str:          # a blocking library you cannot change
    time.sleep(0.2)
    return f"invoice-{order_id}.pdf"

app = FastAPI()

@app.get("/invoices/{order_id}")
async def invoice(order_id: int):
    name = await run_in_threadpool(legacy_pdf_render, order_id)      # off the loop, into the pool
    return {"file": name}

async def main():
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t") as client:
        start = time.perf_counter()
        results = await asyncio.gather(*(client.get(f"/invoices/{i}") for i in range(10)))
        print([r.json()["file"] for r in results][:3], "…", f"10 requests in {time.perf_counter() - start:.2f} s")

asyncio.run(main())

Output

['invoice-0.pdf', 'invoice-1.pdf', 'invoice-2.pdf'] … 10 requests in 0.23 s

4. CPU-bound work: processes, not threads

a process pool created in lifespan, used from async endpoints

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

def score_risk(seed: int) -> int:                     # pure-Python CPU work
    return sum((i * seed) % 7 for i in range(3_000_000))

@asynccontextmanager
async def lifespan(app):
    with ProcessPoolExecutor(max_workers=2) as pool:  # separate processes, separate GILs
        yield {"pool": pool}

app = FastAPI(lifespan=lifespan)

@app.get("/risk/{seed}")
async def risk(seed: int, request: Request):
    loop = asyncio.get_running_loop()
    return {"score": await loop.run_in_executor(request.state.pool, score_risk, seed)}

@app.get("/health")
async def health():
    return "ok"

if __name__ == "__main__":                              # required: pool processes re-import this file
    with TestClient(app) as client, ThreadPoolExecutor(2) as callers:
        client.get("/risk/1")                           # warm up the worker processes
        start = time.perf_counter()
        pending = [callers.submit(client.get, f"/risk/{seed}") for seed in (3, 5)]
        time.sleep(0.05)
        probe = time.perf_counter()
        client.get("/health")                           # while both CPU jobs are running
        waited = time.perf_counter() - probe
        scores = [f.result().json()["score"] for f in pending]
        print(f"scores {scores} in {time.perf_counter() - start:.2f} s | /health answered in {waited * 1000:.0f} ms")

Output

scores [9000000, 8999999] in 0.62 s | /health answered in 2 ms

The CPU work ran in two other processes while the event loop stayed responsive. For heavy or long jobs, go further and move the work out of the web process entirely, into a task queue (part 13), so a spike of expensive requests cannot starve your API.

5. Tuning and gotchas

Recap