FastAPI · Advanced

Async patterns

When to write async def and when plain def, work done after the reply, task queues, WebSockets, and server-sent events.

async

Overview

FastAPI runs on an event loop: a single thread handles many requests by switching between them whenever one is waiting on I/O. That is why async def routes must only await non-blocking calls - one blocking call freezes every request on that worker. Plain def routes are the escape hatch: FastAPI runs them in a thread pool. Work that should happen after the response goes to background tasks, and heavy or must-not-lose work goes to a task queue.

Key concepts

async def: Runs on the event loop; each await hands control back while the call waits.
def: Runs in a thread pool so blocking libraries do not stall the loop. The pool is limited - 40 threads by default.
BackgroundTasks: Run after the response is sent, in the same process - lost if the process restarts.
Task queues: ARQ or Celery run jobs in separate worker processes, with retries and persistence through Redis or another broker.
WebSockets vs SSE: WebSockets are two-way. Server-sent events are one-way, server to client, over plain HTTP, and the browser reconnects automatically.

Best practices

If a library has no async version (requests, many SDKs), call it from a def route or through run_in_threadpool - never directly inside async def.
Use httpx.AsyncClient instead of requests in async code.
Do not use BackgroundTasks for work that must not be lost; use a task queue.

Async patterns

async def vs def - when to use which

Core

async def

I/O you can await: DB queries, HTTP calls, file reads, caching.

def (sync)

Blocking libraries with no async version. FastAPI runs these in a thread pool.

# USE async def when doing awaited I/O: DB, Redis, HTTP calls
@app.get("/users")
async def list_users(db = Depends(get_db)):
    result = await db.execute(select(User))  # awaited I/O
    return result.scalars().all()

# USE def for blocking/sync libraries
# FastAPI runs these in a thread pool automatically
@app.get("/report")
def report():
    return {"result": blocking_library_call()}

# Run blocking code from an async route
from fastapi.concurrency import run_in_threadpool

@app.post("/process")
async def process(data: dict):
    result = await run_in_threadpool(sync_heavy_function, data)
    return result

# NEVER: blocking call inside async def
import time, asyncio

@app.get("/bad")
async def bad():
    time.sleep(5)           # ❌ freezes the event loop - every request waits
    return {}

@app.get("/good")
async def good():
    await asyncio.sleep(5)  # ✅ yields to other requests while waiting
    return {}

Watch out: A thread pool keeps the server free to answer other requests, but heavy number-crunching (image processing, big calculations) still fights for the one lock Python gives to running code. Move that work to a separate process or a task queue.

Background tasks

Background

from fastapi import BackgroundTasks

async def send_welcome_email(email: str):
    # runs AFTER response is returned to client
    await email_service.send(email, "Welcome!")

def log_activity(user_id: int, action: str):
    db.log(user_id, action)

@app.post("/register")
async def register(
    user_data: UserCreate,
    bg: BackgroundTasks
):
    user = await create_user_db(user_data)
    bg.add_task(send_welcome_email, user.email)
    bg.add_task(log_activity, user.id, "registered")
    return user  # returned immediately, tasks run after

Tip: BackgroundTasks run in the same process: if the server restarts, pending tasks are lost. Use a task queue for work that must not be dropped.

Heavy jobs - task queue (ARQ)

Queue

# pip install arq
from arq import create_pool
from arq.connections import RedisSettings

# worker.py - runs in a separate worker process
async def process_video(ctx, video_id: str):
    await encode_video(video_id)

class WorkerSettings:
    functions = [process_video]

# main.py - create the pool once, in lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.arq = await create_pool(RedisSettings())
    yield
    await app.state.arq.close()

@app.post("/videos/{id}/process")
async def queue_job(id: str, request: Request):
    await request.app.state.arq.enqueue_job("process_video", id)
    return {"queued": True}

WebSockets

from fastapi import WebSocket, WebSocketDisconnect

# Connection manager for multiple clients
class ConnectionManager:
    def __init__(self): self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, msg: str):
        for ws in self.active:
            await ws.send_text(msg)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(ws: WebSocket, client_id: str):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)

Tip: This manager only knows about its own process. If you run several worker processes, each keeps its own list of connections - use something like Redis pub/sub to reach them all.

Server-sent events (SSE)

Streaming

from fastapi.responses import StreamingResponse
import asyncio

async def event_stream():
    for i in range(10):
        yield f"data: update {i}\n\n"   # SSE frame: "data: ..." + blank line
        await asyncio.sleep(1)

@app.get("/events")
async def events():
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream"
    )

Comments

Loading comments...

Async patterns

Overview

Key concepts

Best practices

Async patterns

async def vs def - when to use which

Background tasks

Heavy jobs - task queue (ARQ)

WebSockets

Server-sent events (SSE)

Comments

AI

System Design

Backend

Frontend

Fundamentals