FastAPI: sync code in an async route, run_in_threadpool

The single most common production gotcha with FastAPI: calling sync code (DB driver, requests, computation) inside an async def route blocks the event loop, killing concurrency. Two ways to handle it: declare the route sync, or offload explicitly.

What blocks vs what doesn’t

@app.get("/users/{id}")
async def get_user(id: int):
    user = sync_db.query(User).get(id)       # BLOCKS — sync DB call
    return user

While the sync call runs, every other coroutine in the process is frozen. Throughput collapses; tail latency explodes.

Fix 1: Declare the route sync

FastAPI automatically runs def (not async def) routes in a threadpool:

@app.get("/users/{id}")
def get_user(id: int):                       # plain def — runs in threadpool
    user = sync_db.query(User).get(id)
    return user

This is the easy migration path: if you have an existing sync codebase, just declare routes def and FastAPI threadpools them. The event loop stays free; throughput is bounded by threadpool size.

Don’t mix def and async def ad-hoc. Pick deliberately per route based on what work it does.

Fix 2: run_in_threadpool

For when the route must be async def (uses async DB and a synchronous helper inside):

from fastapi.concurrency import run_in_threadpool

@app.get("/users/{id}")
async def get_user(id: int, session: AsyncSession = Depends(get_session)):
    user = await session.get(User, id)                       # async DB
    pdf = await run_in_threadpool(generate_pdf_sync, user)   # sync helper offloaded
    return Response(content=pdf, media_type="application/pdf")

run_in_threadpool is FastAPI’s wrapper around anyio’s to_thread.run_sync — same idea as asyncio.to_thread.

When to use which

Mixing the two in one app is fine — FastAPI handles it. Mixing both in one route is when run_in_threadpool shines.

Threadpool sizing

FastAPI / Starlette use AnyIO’s threadpool. Default size is 40 threads (raised from 32 in older versions). Cap of concurrent sync route handlers + concurrent run_in_threadpool calls.

For sync-heavy apps with many concurrent requests, bump it:

import anyio
anyio.to_thread.current_default_thread_limiter().total_tokens = 100

Or set at startup:

@asynccontextmanager
async def lifespan(app: FastAPI):
    anyio.to_thread.current_default_thread_limiter().total_tokens = 100
    yield

app = FastAPI(lifespan=lifespan)

Symptoms of an undersized threadpool: high p99 latency with low CPU, requests piling up.

Don’t await sync calls

Common confusion. requests.get(...) is sync — you can’t await it. Doing so raises a TypeError or silently breaks. Use httpx.AsyncClient for async HTTP or offload requests.get to a thread.

# WRONG
result = await requests.get(url)             # requests.get is sync; doesn't return awaitable

# RIGHT (async lib)
async with httpx.AsyncClient() as c:
    result = await c.get(url)

# RIGHT (sync lib, offloaded)
result = await run_in_threadpool(requests.get, url)

CPU-bound work

Threadpool helps I/O-bound sync calls; doesn’t help CPU-bound work (GIL). For CPU-bound:

from concurrent.futures import ProcessPoolExecutor
import anyio

executor = ProcessPoolExecutor(max_workers=4)

async def heavy_endpoint():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(executor, cpu_intensive, data)
    return result

Or for truly heavy compute, push to a worker (Celery / Fargate task) and respond async.

Dependencies and sync/async

Depends(...) can be sync or async — FastAPI handles both:

def sync_dependency():           # runs in threadpool
    return get_thing_sync()

async def async_dependency():    # runs in event loop
    return await get_thing_async()

@app.get("/x")
async def endpoint(
    a = Depends(sync_dependency),
    b = Depends(async_dependency),
):
    ...

Behaviour is automatic; pick the one that matches the underlying call.

Common gotchas

• Calling sync DB inside async def without offload. Locks the loop.
• Forgetting to use httpx.AsyncClient and using requests in async routes. Same problem.
• Awaiting a non-coroutine. await sync_function() raises if the result isn’t awaitable.
• Threadpool exhaustion silently queues. Symptom: high latency without obvious CPU usage.
• Mixing async DB driver and sync DB driver in same app. Two pools, double contention. Pick one.
• run_in_threadpool and contextvars. AnyIO propagates context properly; raw loop.run_in_executor does not.

Interview angle

• “You wrote async def get_user and call db.query(User).get(id) inside. What goes wrong?” — db.query is sync; blocks the event loop for the duration. Every other request handler stalls. Either declare the route def (FastAPI threadpools it), use run_in_threadpool, or switch to an async DB driver.
• “What’s the difference between FastAPI’s def and async def route handlers?” — async def runs in the event loop; you must avoid sync I/O. def runs in a threadpool — sync I/O is fine, but you don’t benefit from async concurrency for that route. Both can coexist in one app.
• “What’s run_in_threadpool?” — FastAPI wrapper around AnyIO’s threadpool. Use it inside async def routes when you need to call a sync helper without blocking the loop. Equivalent to asyncio.to_thread.
• “How do you do CPU-bound work in a FastAPI endpoint?” — threadpool doesn’t help (GIL). Use a process pool via loop.run_in_executor(ProcessPoolExecutor, ...), or offload to a worker service (Celery, Fargate).
• “What’s the default threadpool size and why does it matter?” — 40 in modern FastAPI/AnyIO. Caps concurrent sync route handlers + run_in_threadpool calls. Under-sized → requests queue, p99 latency rises with low CPU.
• “How would you migrate a sync FastAPI app to fully async?” — switch DB driver to async (SQLAlchemy 2.0 async + asyncpg), switch HTTP to httpx, convert routes from def to async def, replace remaining sync helpers with run_in_threadpool until they’re rewritten async.