FastAPI Async: Event Loop, Background Tasks, WebSockets

FastAPI is the only major Python framework where the event loop is the default, not an opt-in. Starlette and asyncio do the heavy lifting underneath — your job is to know which endpoints should be async def, which should stay def, and which library calls inside an async route will silently freeze every other request on the worker.

Run locally with pip install 'fastapi[standard]' httpx and fastapi dev main.py. Expected output shown in comments.

This lesson builds on async — coroutines, await, asyncio.gather. Skim that if any of it is fuzzy.

1. Why FastAPI Is Async-Native

Flask runs your view function on a worker thread. The worker is blocked for the full duration of the request — DB queries, HTTP calls, response serialisation. Concurrency comes from spawning more workers (processes or threads), each one expensive.

FastAPI runs on Starlette, which runs on asyncio. One worker can hold thousands of in-flight requests as long as each one spends most of its life awaiting I/O. A 500 ms downstream API call doesn't block anything — the loop just runs other coroutines until the response arrives. This is the same model behind Node.js and Go, brought to Python via async/await.

The catch — async only works when you actually await. CPU work, blocking libraries, and time.sleep inside an async def route freeze the event loop and tank throughput. Knowing the rule is the entire skill.

2. `async def` vs `def` — The Decision Rule

FastAPI accepts both. The difference is where the function runs.

You write	Runs on	Good for
`async def`	The event loop, in the main thread	Code that `await`s async libraries
`def`	A worker threadpool (default 40 threads)	Blocking libraries you can't replace

Two rules:

Use async def when your handler awaits something — async DB driver, httpx.AsyncClient, aiofiles, redis.asyncio.
Use def when your handler calls blocking libraries you can't change — requests, the standard sqlite3, psycopg2, a slow C extension. FastAPI will run it on a threadpool so the event loop stays free.

The worst combination is async def with blocking calls inside:

python

import time, requests

@app.get("/bad")
async def bad():
    time.sleep(1)                       # blocks the event loop for 1s
    r = requests.get("https://api.example.com")
    return r.json()

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

Every other request on this worker waits. Throughput collapses from "thousands per second" to "one per second". If your route is async def, every blocking line in it has to be either replaced with an async equivalent or pushed to a thread with await asyncio.to_thread(blocking_fn, ...).

3. The Canonical Async Endpoint

python

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/joke")
async def get_joke():
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get("https://api.chucknorris.io/jokes/random")
        r.raise_for_status()
        return r.json()
# GET /joke -> {"id":"...", "value":"Chuck Norris..."}

+ setup added so this can run · defines client

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

client = _AutoMock('client')

async def, await client.get(...), async with for the client — the loop stays free while we wait for the upstream response. Run two hundred concurrent requests against this endpoint on one worker — they all overlap.

The connection pooling lives inside the client. Building a fresh AsyncClient per request is correct here but wasteful — at scale, lift it to a module-level singleton, opened in the lifespan and reused across handlers.

4. Concurrent Dependent Calls with `asyncio.gather`

A page that needs three upstream calls — user profile, recent posts, follower count — should fire them concurrently, not sequentially. Three 200 ms calls in series take 600 ms; in parallel they take ~200.

python

import asyncio, httpx

async def fetch_json(client, url):
    r = await client.get(url, timeout=5.0)
    r.raise_for_status()
    return r.json()

@app.get("/users/{user_id}/dashboard")
async def dashboard(user_id: int):
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        user, posts, followers = await asyncio.gather(
            fetch_json(client, f"/users/{user_id}"),
            fetch_json(client, f"/users/{user_id}/posts"),
            fetch_json(client, f"/users/{user_id}/followers/count"),
        )
    return {"user": user, "posts": posts, "follower_count": followers}

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

asyncio.gather schedules all three coroutines on the loop and waits for them together. Results come back in input order. If any one raises, the exception propagates from gather (the others keep running by default — pass return_exceptions=True to collect errors instead). On Python 3.11+, asyncio.TaskGroup is the modern replacement with cleaner cancellation semantics.

5. Async Databases

The async ecosystem now has a first-class story for every major database. Two examples — pick whichever fits the project.

asyncpg for PostgreSQL — the fast, focused driver:

python

import asyncpg
from fastapi import FastAPI

app = FastAPI()
pool: asyncpg.Pool

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, name, email FROM users WHERE id = $1", user_id
        )
        if row is None:
            raise HTTPException(404, "Not found")
        return dict(row)

+ setup added so this can run · defines HTTPException, conn

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def HTTPException(*_a, **_kw):
    print('-> HTTPException() called')
    return _AutoMock('HTTPException()')
conn = _AutoMock('conn')

SQLAlchemy 2.0 in async mode — ORM with async semantics:

python

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pw@host/db")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    async with SessionLocal() as session:
        item = await session.get(Item, item_id)
        if item is None:
            raise HTTPException(404, "Not found")
        return item

+ setup added so this can run · defines app, Item, HTTPException, session

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
Item = _AutoMock('Item')
def HTTPException(*_a, **_kw):
    print('-> HTTPException() called')
    return _AutoMock('HTTPException()')
session = _AutoMock('session')

The pool/engine is shared across the application — open it once at startup (lifespan handles this), inject the session via a dependency per request.

The blocking equivalents — psycopg2, sync SQLAlchemy — work too, but only inside def routes (so FastAPI threadpools them) or via await asyncio.to_thread(...). Mixing a blocking driver into an async def handler is the single most common FastAPI performance bug.

6. Background Tasks — Fire-and-Forget After the Response

Sometimes you want to return immediately, then do work — log an event, send a welcome email, write to an analytics queue. BackgroundTasks runs the work after the response is sent.

python

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("audit.log", "a") as f:
        f.write(message + "\n")

async def send_welcome_email(email: str):
    await email_client.send(email, subject="Welcome", body="...")

@app.post("/signup")
async def signup(email: str, background: BackgroundTasks):
    user = await create_user(email)
    background.add_task(write_log, f"signup {email}")
    background.add_task(send_welcome_email, email)
    return {"id": user.id, "email": email}

+ setup added so this can run · defines app, create_user, email_client

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
def create_user(*_a, **_kw):
    print('-> create_user() called')
    return _AutoMock('create_user()')
email_client = _AutoMock('email_client')

The handler returns. Then the background tasks run, in order, on the same worker. Both sync (def) and async (async def) tasks are supported.

When to use it — short, best-effort work where a failure shouldn't bubble up to the client. When not — long-running jobs, anything that must retry, anything you want to monitor or scale independently. For that, push to a real queue (Redis + RQ, Celery, NATS, or a cloud queue) and have a worker process consume it. BackgroundTasks runs in-process — if the worker dies after responding but before the task finishes, the work is lost.

7. Lifespan — Startup and Shutdown the Modern Way

Pre-warming an ML model, opening a DB pool, starting a background scheduler — you need code that runs once when the app starts and once when it shuts down. The old @app.on_event("startup") / "shutdown" decorators are deprecated. Use lifespan:

python

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    app.state.db = await asyncpg.create_pool(
        dsn="postgresql://user:pw@localhost/db",
        min_size=5, max_size=20,
    )
    print("DB pool opened")
    yield
    # Shutdown
    await app.state.db.close()
    print("DB pool closed")

app = FastAPI(lifespan=lifespan)

@app.get("/health")
async def health():
    async with app.state.db.acquire() as conn:
        await conn.execute("SELECT 1")
    return {"ok": True}

+ setup added so this can run · defines conn

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

conn = _AutoMock('conn')

The function is an async context manager — code before yield runs at startup, code after runs at shutdown. The yielded object is available on app.state (or you can stash things on app.state.x directly). This is where httpx.AsyncClient, DB pools, Redis clients, ML model loaders, and background scheduler threads should live.

Why this matters — without lifespan, you'd open a new DB connection on every request (slow) or open a pool at module-import time (untestable, leaks on shutdown). Lifespan gives you a single bounded scope that mirrors the app's lifetime.

8. Streaming Responses

For large payloads, server-sent events, or anything you don't want to fit in memory, return a StreamingResponse over a generator.

python

from fastapi.responses import StreamingResponse
import asyncio

async def number_stream():
    for i in range(10):
        await asyncio.sleep(0.5)
        yield f"data: {i}\n\n"             # SSE wire format

@app.get("/events")
async def events():
    return StreamingResponse(
        number_stream(),
        media_type="text/event-stream",
    )
# curl http://127.0.0.1:8000/events
# data: 0
# data: 1
# data: 2
# ...

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

The client receives bytes as the generator produces them — the response is not buffered to completion. Use this for CSV exports of millions of rows, model token streams from LLMs, progress updates, or anything pull-based from a downstream system. For SSE specifically, set media_type="text/event-stream" and format chunks as data: ...\n\n.

For a large file, FileResponse("path") is simpler — Starlette handles the chunked read.

9. WebSockets — Bidirectional, Long-Lived

HTTP is request/response. WebSockets are a persistent two-way connection — once upgraded, either side can send a message at any time. FastAPI exposes them with @app.websocket("/path"):

python

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/echo")
async def echo(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            msg = await ws.receive_text()
            await ws.send_text(f"echo: {msg}")
    except WebSocketDisconnect:
        print("client disconnected")

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

accept() finishes the HTTP-to-WS upgrade handshake. receive_text()/receive_bytes()/receive_json() await a frame from the client. send_text() etc. push a frame back. The connection lives until the client disconnects or you await ws.close().

Real apps need a ConnectionManager to broadcast to all clients, handle reconnects, and clean up on disconnect — the FastAPI docs have the canonical chat-room recipe. WebSockets are perfect for chat, live dashboards, multiplayer games, and anywhere you'd otherwise reach for long-polling.

10. The `Annotated` Pattern — Modern Parameter Metadata

FastAPI 0.95+ prefers Annotated[Type, Marker(...)] over Type = Marker(default, ...). Both still work, but Annotated plays better with type checkers, default values, and dependency injection:

python

from typing import Annotated
from fastapi import Path, Query, Header

@app.get("/items/{item_id}")
def get_item(
    item_id: Annotated[int, Path(ge=1, le=10_000)],
    q: Annotated[str | None, Query(min_length=2, max_length=50)] = None,
    user_agent: Annotated[str | None, Header()] = None,
):
    ...

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

The type signature reads as Python: item_id is an int, q is an str or None. The Path/Query/Header marker carries FastAPI metadata. Default values stay where Python expects them — after the =, not buried in the marker. The same shape extends to Depends(...) (see fastapi-deps) and any custom validator.

11. CPU-Bound Work — Off the Loop

asyncio runs on one thread. A heavy synchronous computation inside async def freezes everything. Push it to a thread pool (cheap, shares memory) or process pool (bypasses the GIL, separate processes):

python

import asyncio, hashlib

@app.post("/hash")
async def hash_bytes(payload: bytes = Body(...)):
    # 50 MB SHA-256 — CPU-bound, would block the loop
    digest = await asyncio.to_thread(
        hashlib.sha256(payload).hexdigest
    )
    return {"digest": digest}

+ setup added so this can run · defines Body, app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def Body(*_a, **_kw):
    print('-> Body() called')
    return _AutoMock('Body()')
app = _AutoMock('app')

asyncio.to_thread(fn, *args) (3.9+) is the one-liner. For process-pool parallelism — multiple cores, no GIL — use loop.run_in_executor(ProcessPoolExecutor(), fn, *args). Either way, the route stays async def and the loop stays responsive.

Common Mistakes

1. time.sleep in an async def route

python

@app.get("/wait")
async def wait():
    time.sleep(5)                       # FREEZES the entire worker for 5s
    return {"done": True}

+ setup added so this can run · defines app, time

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
time = _AutoMock('time')

Every other in-flight request waits. The fix is await asyncio.sleep(5). The same bug shows up with requests.get, blocking DB drivers, urllib, file I/O without aiofiles. If it's not awaitable, it doesn't belong in an async def handler.

2. Using requests instead of httpx.AsyncClient

python

import requests

@app.get("/proxy")
async def proxy():
    return requests.get("https://api.example.com").json()

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

Same bug as above. requests is synchronous. Use httpx.AsyncClient for outbound HTTP from async routes. (Or downgrade the route to def — FastAPI threadpools it for you. But for anything I/O-heavy, do the rewrite.)

3. Long-running CPU work inside async def

python

@app.post("/process")
async def process(data: bytes):
    return heavy_image_filter(data)     # 3 seconds of NumPy — loop frozen

+ setup added so this can run · defines heavy_image_filter, app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def heavy_image_filter(*_a, **_kw):
    print('-> heavy_image_filter() called')
    return _AutoMock('heavy_image_filter()')
app = _AutoMock('app')

Wrap with await asyncio.to_thread(heavy_image_filter, data) for thread-pool offload, or loop.run_in_executor(process_pool, heavy_image_filter, data) for true parallelism. The handler should never be the CPU-bound work.

4. Never closing the connection pool

Opening a DB pool at module import time means it never gets a chance to close cleanly on shutdown. Connections leak, integration tests warn. Use lifespan for any resource with a non-trivial lifecycle — DB pools, HTTP clients, background scheduler threads.

5. Treating BackgroundTasks as a job queue

BackgroundTasks runs in the same process, in memory, after the response. A worker restart loses pending tasks. No retries, no priorities, no observability. For anything that must complete, push to a real queue. For "log this, kick off an email" — fine.

6. asyncio.run inside a route

python

@app.get("/oops")
async def oops():
    asyncio.run(other_coro())           # RuntimeError — a loop is already running

+ setup added so this can run · defines app, asyncio, other_coro

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
asyncio = _AutoMock('asyncio')
def other_coro(*_a, **_kw):
    print('-> other_coro() called')
    return _AutoMock('other_coro()')

You're already inside the event loop. Just await other_coro(). asyncio.run is for the very top of a synchronous program — not inside a coroutine, and definitely not inside a FastAPI handler.

🎯 Your Turn — Concurrent Multi-API Aggregator

Write GET /aggregate?user_id=... that hits three (fake) upstream APIs concurrently with httpx.AsyncClient and asyncio.gather, then returns a combined JSON object. Use httpbin.org endpoints so it runs without setup, with timeouts so one slow service doesn't sink the request.

Requirements:

async def route, httpx.AsyncClient with timeout=3.0.
Three URLs hit concurrently with asyncio.gather.
A 504 HTTPException if the whole thing exceeds 5 seconds (use asyncio.timeout).
Per-call failure tolerance — if one upstream returns 500, the response should include that service's slot as null, not 500 the whole route.

python

# main.py
import asyncio
import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()

URLS = {
    "ip": "https://httpbin.org/ip",
    "ua": "https://httpbin.org/user-agent",
    "uuid": "https://httpbin.org/uuid",
}

# TODO 1: define `fetch_one(client, name, url)` that returns (name, dict | None)
# TODO 2: define GET /aggregate that:
#          - opens an AsyncClient
#          - uses asyncio.timeout(5.0) as the overall budget
#          - calls asyncio.gather(*[fetch_one(client, n, u) for n, u in URLS.items()])
#          - returns dict(results)
# TODO 3: catch TimeoutError -> HTTPException(504)

Hint 1 — Per-call try/except

Wrap the client.get(...) inside fetch_one with a try/except that catches httpx.HTTPError and returns (name, None). Without that, one failing upstream propagates through gather and kills the whole batch.

Hint 2 — The 3.11+ timeout context

asyncio.timeout(5.0) is the modern way to bound a block: async with asyncio.timeout(5.0): result = await gather(...). A TimeoutError is raised on overrun. Catch it at the route boundary and translate to a 504 via HTTPException.

Show full solution

python

# main.py
import asyncio
import httpx
from fastapi import FastAPI, HTTPException, status

app = FastAPI()

URLS = {
    "ip": "https://httpbin.org/ip",
    "ua": "https://httpbin.org/user-agent",
    "uuid": "https://httpbin.org/uuid",
}


async def fetch_one(client: httpx.AsyncClient, name: str, url: str):
    """Fetch one upstream — return (name, body) or (name, None) on failure."""
    try:
        r = await client.get(url, timeout=3.0)
        r.raise_for_status()
        return name, r.json()
    except httpx.HTTPError:
        return name, None


@app.get("/aggregate")
async def aggregate():
    try:
        async with asyncio.timeout(5.0):
            async with httpx.AsyncClient() as client:
                results = await asyncio.gather(
                    *[fetch_one(client, n, u) for n, u in URLS.items()]
                )
    except TimeoutError:
        raise HTTPException(
            status_code=status.HTTP_504_GATEWAY_TIMEOUT,
            detail="Upstream aggregation exceeded 5s budget",
        )
    return dict(results)

# Sample response:
# {
#   "ip":   {"origin": "203.0.113.42"},
#   "ua":   {"user-agent": "python-httpx/0.27.0"},
#   "uuid": {"uuid": "8d3e..."}
# }

The pattern in this solution scales to any aggregation route — dashboard endpoints, recommendation feeds, search-results pages — wherever the response is built from multiple upstream calls. The four moving parts are:

One shared AsyncClient — reuses the connection pool across the fan-out. Building a new client per inner call would defeat the optimisation.
Per-call try/except — each upstream's failure becomes a None slot rather than a 500 on the whole route. The response shape stays predictable.
Outer asyncio.timeout — a hard ceiling on the whole block. Without it a degraded upstream can keep your route hung past any sensible client deadline.
asyncio.gather — concurrent dispatch, in-order results, one await for the whole batch.

For very large fan-outs (50+ upstreams), add a Semaphore to cap in-flight requests — same pattern as the bounded fetcher in async. For Python 3.11+ codebases, replace gather with asyncio.TaskGroup so any unexpected exception cancels its siblings.

What You Learned

FastAPI runs on Starlette + asyncio. Async is the default, not an opt-in.
async def runs on the event loop; def runs on a threadpool. async def + a blocking call freezes the loop — the canonical FastAPI bug.
Use httpx.AsyncClient, asyncpg, redis.asyncio, aiofiles — async-native libraries — inside async def routes.
asyncio.gather runs multiple coroutines concurrently in one handler. TaskGroup (3.11+) is the modern replacement.
BackgroundTasks runs short, best-effort work after the response. Real queues for everything else.
Lifespan (@asynccontextmanager async def lifespan(app):) replaces deprecated startup/shutdown events. Open DB pools and HTTP clients there.
StreamingResponse streams large payloads; SSE is one line of media type.
@app.websocket gives you persistent bidirectional channels.
CPU-bound code in async routes → await asyncio.to_thread(...) or a process pool. Never run heavy NumPy/regex/hashing inline.
Annotated[Type, Marker(...)] is the modern syntax for Path/Query/Header/Depends metadata.

Next: FastAPI Dependency Injection — the killer feature you've been hinting at: pluggable, composable, testable building blocks for auth, DB sessions, settings, and route protection.

FastAPI Async: Event Loop, Background Tasks, WebSockets

1. Why FastAPI Is Async-Native

2. async def vs def — The Decision Rule

3. The Canonical Async Endpoint

4. Concurrent Dependent Calls with asyncio.gather

5. Async Databases

6. Background Tasks — Fire-and-Forget After the Response

7. Lifespan — Startup and Shutdown the Modern Way

8. Streaming Responses

9. WebSockets — Bidirectional, Long-Lived

10. The Annotated Pattern — Modern Parameter Metadata

11. CPU-Bound Work — Off the Loop

Common Mistakes

🎯 Your Turn — Concurrent Multi-API Aggregator

What You Learned

2. `async def` vs `def` — The Decision Rule

4. Concurrent Dependent Calls with `asyncio.gather`

10. The `Annotated` Pattern — Modern Parameter Metadata