async-fastapi

Async FastAPI

Build high-performance async APIs with FastAPI. Async endpoints handle concurrent requests efficiently without blocking the event loop, which is critical for I/O-bound workloads like database queries, HTTP calls, and file operations.

When to Use

• User creates or modifies FastAPI endpoints
• User needs concurrent I/O operations
• User implements WebSockets or streaming
• User asks about background processing
• User encounters event loop blocking or performance issues

Core Patterns

Async Endpoints

Use async def for I/O-bound endpoints. Use plain def for CPU-bound work (FastAPI runs sync handlers in a threadpool automatically).

from fastapi import FastAPI
import httpx

app = FastAPI()

# Async -- for I/O-bound operations (DB, HTTP, file)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

# Sync -- for CPU-bound operations (FastAPI runs in threadpool)
@app.get("/compute/{n}")
def compute_heavy(n: int):
    return {"result": sum(i * i for i in range(n))}

# Concurrent async operations
import asyncio

@app.get("/dashboard/{user_id}")
async def get_dashboard(user_id: int):
    user_task = get_user_from_db(user_id)
    orders_task = get_orders_from_db(user_id)
    notifications_task = get_notifications(user_id)

    user, orders, notifications = await asyncio.gather(
        user_task, orders_task, notifications_task
    )

    return {"user": user, "orders": orders, "notifications": notifications}

Lifespan Events

Use the lifespan context manager to handle startup and shutdown. This replaces the deprecated on_event decorators.

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize shared resources
    app.state.http_client = httpx.AsyncClient(timeout=30.0)
    app.state.db_pool = await create_db_pool()
    yield
    # Shutdown: clean up resources
    await app.state.http_client.aclose()
    await app.state.db_pool.close()

app = FastAPI(lifespan=lifespan)

@app.get("/fetch")
async def fetch_data(url: str):
    response = await app.state.http_client.get(url)
    return response.json()