Python Async Patterns with FastAPI

FastAPI is built on Python's async/await, enabling high-performance APIs. This guide covers advanced async patterns beyond basic CRUD.

Background Tasks

from fastapi import FastAPI, BackgroundTasks
from fastapi_mail import FastMail, MessageSchema, ConnectionConfig

app = FastAPI()

# Simple background task
def send_email_sync(email: str, message: str):
    # This runs after response is sent
    mail_client.send(email, message)

@app.post("/register")
async def register(user: UserCreate, background_tasks: BackgroundTasks):
    db_user = await create_user(user)
    
    # Add to background queue (non-blocking)
    background_tasks.add_task(send_welcome_email, db_user.email)
    background_tasks.add_task(update_analytics, "user_registered")
    
    return {"id": db_user.id, "status": "created"}

# For heavy tasks, use Celery or arq
import arq

async def send_welcome_email(ctx, email: str):
    """Runs in arq worker"""
    await email_service.send_welcome(email)

class WorkerSettings:
    functions = [send_welcome_email]
    redis_settings = arq.connections.RedisSettings(host='redis')

@app.post("/register-heavy")
async def register_heavy(user: UserCreate):
    db_user = await create_user(user)
    
    redis = await arq.create_pool(arq.connections.RedisSettings())
    await redis.enqueue_job('send_welcome_email', db_user.email)
    
    return {"id": db_user.id}

WebSocket Connections

from fastapi import WebSocket, WebSocketDisconnect
from typing import Dict, Set

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[str, Set[WebSocket]] = {}
    
    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.active_connections:
            self.active_connections[room] = set()
        self.active_connections[room].add(websocket)
    
    async def disconnect(self, websocket: WebSocket, room: str):
        self.active_connections[room].discard(websocket)
    
    async def broadcast(self, message: dict, room: str):
        if room not in self.active_connections:
            return
        dead = set()
        for ws in self.active_connections[room]:
            try:
                await ws.send_json(message)
            except Exception:
                dead.add(ws)
        self.active_connections[room] -= dead

manager = ConnectionManager()

@app.websocket("/ws/{room}/{user_id}")
async def websocket_endpoint(
    websocket: WebSocket,
    room: str,
    user_id: str,
):
    await manager.connect(websocket, room)
    try:
        while True:
            data = await websocket.receive_json()
            await manager.broadcast({
                "from": user_id,
                "message": data["message"],
                "timestamp": datetime.utcnow().isoformat(),
            }, room)
    except WebSocketDisconnect:
        await manager.disconnect(websocket, room)
        await manager.broadcast({"system": f"{user_id} left"}, room)

Server-Sent Events (SSE)

from fastapi.responses import StreamingResponse
import asyncio

@app.get("/events/{user_id}")
async def sse_events(user_id: str, request: Request):
    async def event_generator():
        try:
            async for event in event_stream(user_id):
                if await request.is_disconnected():
                    break
                yield f"data: {json.dumps(event)}

"
                await asyncio.sleep(0)  # Yield control
        except asyncio.CancelledError:
            pass

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Disable nginx buffering
        },
    )

# Streaming LLM responses
@app.post("/chat/stream")
async def chat_stream(message: ChatMessage):
    async def generate():
        async with openai.AsyncOpenAI() as client:
            stream = await client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": message.content}],
                stream=True,
            )
            async for chunk in stream:
                if chunk.choices[0].delta.content:
                    yield f"data: {chunk.choices[0].delta.content}

"
        yield "data: [DONE]

"

    return StreamingResponse(generate(), media_type="text/event-stream")

Async Database with SQLAlchemy

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(User).where(User.id == user_id).options(
            selectinload(User.orders)  # Eagerly load related
        )
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "User not found")
    return user

# Concurrent DB queries
@app.get("/dashboard/{user_id}")
async def get_dashboard(user_id: int, db: AsyncSession = Depends(get_db)):
    # Run queries concurrently
    user_task = asyncio.create_task(get_user_by_id(db, user_id))
    stats_task = asyncio.create_task(get_user_stats(db, user_id))
    notifications_task = asyncio.create_task(get_notifications(db, user_id))
    
    user, stats, notifications = await asyncio.gather(
        user_task, stats_task, notifications_task
    )
    
    return {"user": user, "stats": stats, "notifications": notifications}

Middleware and Lifespan

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    app.state.redis = await aioredis.create_redis_pool("redis://localhost")
    app.state.http_client = httpx.AsyncClient()
    print("App started")
    
    yield
    
    # Shutdown
    app.state.redis.close()
    await app.state.redis.wait_closed()
    await app.state.http_client.aclose()
    print("App shutdown")

app = FastAPI(lifespan=lifespan)

# Request timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Summary

FastAPI async patterns:

• Use BackgroundTasks for lightweight post-response work
• Use Celery/arq for CPU-intensive or long-running tasks
• WebSocket ConnectionManager for room-based broadcasting
• SSE for one-way server-to-client streaming (LLM responses)
• asyncio.gather for concurrent database queries
• Lifespan context manager for startup/shutdown resources