Async and Background Work

Declare endpoints async def when they await async libraries (asyncpg, httpx, aioredis). If you must call blocking code (requests, sync SQLAlchemy, CPU-heavy work), either declare the endpoint with plain def (FastAPI runs it in a threadpool) or use run_in_threadpool/anyio.to_thread. Blocking calls inside async def freeze the event loop for every request. For quick post-response work use BackgroundTasks; for durable, retryable or heavy jobs use a queue (Celery, ARQ, Dramatiq).

import asyncio
from fastapi import BackgroundTasks
from starlette.concurrency import run_in_threadpool


@app.get("/reports/{id}")
async def report(id: int, http: HTTP):
    # Concurrent I/O: both requests in flight at once
    summary, details = await asyncio.gather(
        http.get(f"https://svc/summary/{id}"),
        http.get(f"https://svc/details/{id}"),
    )
    pdf = await run_in_threadpool(render_pdf_sync, summary.json(), details.json())   # blocking lib
    return Response(pdf, media_type="application/pdf")


@app.post("/signup", status_code=201)
async def signup(body: SignupIn, background: BackgroundTasks, db: DB):
    user = await users.create(db, body)
    background.add_task(send_welcome_email, user.email)   # runs after the response
    return {"id": user.id}

Common pitfalls

• Calling requests.get() or time.sleep() inside async def.
• Using BackgroundTasks for critical work that must survive restarts.
• Unbounded asyncio.gather over thousands of calls (use a semaphore).

Key points

• Never block the event loop
• def endpoints run in a threadpool
• Queues for durable background jobs

Further reading: FastAPI: Concurrency and async/await

Go deeper: WSGI, ASGI and the Event Loop in the Request Lifecycle course traces this step of the request, with a production bug and a runnable simulation.