Home / Guides / FastAPI async blocking

Don't put blocking calls inside FastAPI async def

Agents love async def. It looks modern, and FastAPI examples use it. Then they call requests.get, time.sleep, or session.query(User) inside that coroutine. Those calls never yield. One slow dependency freezes every other request on that worker.

FastAPI’s own async docs say: if the library supports await, use async def; if it only blocks, use plain def. This page is that one rule, with the mistakes agents actually ship. For ownership checks, Alembic and pytest on FastAPI + SQLAlchemy 2, see AGENTS.md and Cursor rules for FastAPI. For a venv / pytest skeleton without FastAPI, see Cursor rules for Python.

What “blocking” means here

An async def path operation runs on the event loop. await is the only way that coroutine gives the loop back. A function that sits in C or in a socket recv without an event-loop integration holds the loop until it returns. During that time the worker cannot accept other connections, cannot run other coroutines, and cannot flush background tasks.

Plain def path operations are different. FastAPI runs those in a threadpool and awaits the thread. Dependencies declared as def get the same treatment. Helpers you call from async def do not. That last sentence is the one agents miss.

The usual blocking calls

import time
import requests
from sqlalchemy.orm import Session

@app.get("/slow")
async def slow(db: Session):
    time.sleep(2)                          # blocks the loop
    r = requests.get("https://example.com")  # blocks the loop
    user = db.query(User).get(1)           # sync ORM, blocks
    return {"ok": r.ok, "user": user.id}

Replace each one with something that yields, or stop claiming the route is async.

import asyncio
import httpx
from fastapi.concurrency import run_in_threadpool
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

@app.get("/ok")
async def ok(db: AsyncSession):
    await asyncio.sleep(2)
    async with httpx.AsyncClient() as client:
        r = await client.get("https://example.com")
    user = (await db.execute(select(User).where(User.id == 1))).scalar_one()
    return {"ok": r.is_success, "user": user.id}

@app.get("/legacy-sync")
def legacy_sync(db: Session):
    # FastAPI runs this def route in a threadpool.
    return db.query(User).get(1)

@app.get("/must-stay-async")
async def must_stay_async():
    return await run_in_threadpool(legacy_report)

Use httpx.AsyncClient (or any client whose methods are await-able), asyncio.sleep, and SQLAlchemy 2 AsyncSession with an async driver such as asyncpg. If a library has no async API, either declare the route as def or wrap the call in run_in_threadpool.

SQLAlchemy specifics agents get wrong

• session.query() is the 1.x API and it is synchronous. In a 2.x codebase the rule is select() + await session.execute on an AsyncSession.
• Lazy loads inside async def issue implicit sync IO when you touch a relationship. That is a blocking call you cannot see on the line you wrote. Use selectinload / await session.refresh instead.
• Do not share one sync Session across async tasks. Thread-local sync sessions and the event loop do not mix. One async session per request, closed in a dependency.

If the app is not on async SQLAlchemy yet, it is valid to keep the routes as def. That is what FastAPI recommends when the library cannot be awaited. Do not sprinkle async def on top of a sync session “for speed”.

A rule the agent can follow

---
description: No blocking IO in async def
globs: app/**/*.py
alwaysApply: false
---
- Inside async def, only call await-able IO. No time.sleep, requests, boto3 sync, httpx.Client, or session.query().
- HTTP: httpx.AsyncClient (or equivalent) and await it.
- Sleep: asyncio.sleep. ORM: AsyncSession + select(); no lazy loads.
- If a library is sync-only: declare the path operation as def, or await run_in_threadpool(sync_fn).
- def helpers you call from async def are not moved to a threadpool. Wrap them.

Add the same four bullets to AGENTS.md so Codex and Claude Code see them. Pair with the existing FastAPI ownership rule: authorization still belongs in the query, owner_id == user.id, other users’ objects return 404. Blocking IO and IDOR are separate bugs; agents often introduce both in one endpoint.

How to catch it in review

Grep is enough for a first pass:

rg -n "async def" -g 'app/**/*.py'
rg -n "time\\.sleep|requests\\.|httpx\\.Client\\(|\\.query\\(|boto3\\.|subprocess\\.run" -g 'app/**/*.py'

If both match in the same function, the agent blocked the loop. Also search for open( on large files and json.load of a remote download inside async def. CPU-heavy work (image resize, PDF render) belongs in a thread or a worker, not on the loop.

A useful pytest is concurrency, not just status codes: fire 20 overlapping requests at a route that “sleeps” 200ms. If wall time is ~4 seconds, you serialized on the loop. If it is ~200–400ms, the sleeps overlapped. You do not need a benchmark suite; one overlapping-request test is enough to fail a PR.

When async def is the right default

Use it when the route awaits an async engine, an async HTTP client, or several of those concurrently with asyncio.gather. FastAPI will still serve def routes quickly for typical apps. Tiangolo’s guidance is: if you are not sure, use def. Agents invert that and stamp async def on every handler. The rule is not “never async”. The rule is “do not lie to the event loop”.

Background tasks have the same trap. A BackgroundTasks function that is def is fine. An async def background task that calls requests blocks the loop after the response is sent, which still stalls the worker. Wrap or rewrite those too.

FAQ