LLM Best Practices

FastAPI: Async I/O

Last updated 3 min read

View markdown source

AI agents: read this first: llmbestpractices.com/start-here

This is LLM Best Practices (llmbestpractices.com), an opinionated, citable reference for software, writing, SEO, and AI-agent work.

  1. Route, do not crawl. Fetch llmbestpractices.com/llms.txt and open only the pages whose one-line summary matches your task.
  2. Read raw. Append .md to any page URL for markdown. Check status and last_updated in the frontmatter, then read the rules.
  3. Apply as defaults. First-party docs and the project's own conventions win on conflict. Warn before relying on a fast-moving page older than 12 months.
  4. Cite. Link the page by title and URL, e.g. Python, with last_updated for time-sensitive rules. License CC BY 4.0.

Overview

An async def route runs on the event loop, so one blocking call stalls every in-flight request on that worker until it returns. A plain def route runs in a threadpool instead. Choosing between them correctly, and sizing workers and pools, is the performance contract of FastAPI. The general asyncio rules are in python-async.

Match the route type to what the route calls

  • async def: every I/O call inside is awaited (asyncpg, httpx.AsyncClient, async SQLAlchemy).
  • def: the route calls a blocking library (a sync SDK, requests, sync SQLAlchemy). FastAPI runs it in a threadpool, so the event loop stays free. This is the correct choice for blocking code, not a workaround.
  • Never call time.sleep, requests.get, or sync file and database I/O inside async def. Use await asyncio.sleep() and httpx.AsyncClient.
@app.get("/users/{user_id}")
async def get_user(user_id: str, db: DB) -> UserRead:
    return await user_repo.get(db, user_id)

Know the threadpool has 40 threads

def routes and dependencies run through AnyIO’s default limiter, which allows 40 concurrent threads. Past that, requests queue. Offload blocking work from async def code with starlette.concurrency.run_in_threadpool or anyio.to_thread.run_sync; both share that same 40-thread limiter and its accounting. asyncio.to_thread uses a separate default executor, so mixing it in splits your thread budget.

from starlette.concurrency import run_in_threadpool
 
async def resize(data: bytes) -> bytes:
    return await run_in_threadpool(_resize_sync, data)

Threads do not parallelize CPU-bound Python because of the GIL. Send CPU-heavy work to a process pool or a worker queue (fastapi-background-tasks).

Use async database drivers exclusively in async routes

For postgres, use asyncpg, psycopg in async mode, or async SQLAlchemy on top of either. Do not use psycopg2 or sync SQLAlchemy in async def.

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", pool_size=10, max_overflow=20)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

Set expire_on_commit=False: after a commit, attribute access on an expired object triggers a lazy load, which fails in async code. Create the engine and one shared httpx.AsyncClient in lifespan; see fastapi-dependencies.

Size workers and pools together

One uvicorn process uses one CPU core. Each worker has its own event loop and its own pool, so the database sees workers x (pool_size + max_overflow) connections. With pool_size=10, max_overflow=20, and four workers, that is up to 120. Put PgBouncer in front when the product exceeds max_connections.

fastapi run --workers 4 app/main.py        # or: uvicorn app.main:app --workers 4
gunicorn app.main:app -k uvicorn_worker.UvicornWorker -w 4   # needs the uvicorn-worker package

On Kubernetes or another orchestrator, FastAPI’s docs recommend one process per container and scaling by replicas rather than workers. The built-in uvicorn.workers module is deprecated in favor of the separate uvicorn-worker package.

Cap concurrency to downstream services

Async code can open far more simultaneous requests to a dependency than threads ever did. Bound them with a semaphore, one per downstream service so each limit tunes independently.

_sem = asyncio.Semaphore(50)
 
async def call_downstream(client: httpx.AsyncClient, url: str) -> dict:
    async with _sem:
        r = await client.get(url, timeout=10.0)
        r.raise_for_status()
        return r.json()

Create the semaphore at module scope or in lifespan. Profile async bottlenecks with python-performance.

Backlinks