FastAPI Async: Event Loop, Background Tasks, WebSockets

In FastAPI, the event loop is the default, not an option you have to turn on. Starlette and asyncio do most of the work. Your job is to know which endpoints should be async def, which should stay def, and which library calls inside an async route will freeze every other request on the worker, without any error.

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

This lesson builds on async (coroutines, await, asyncio.gather). Read that lesson again if any of these are unclear.

1. Why FastAPI Is Async-Native

Flask runs your view function on a worker thread. The worker is blocked for the whole request: DB queries, HTTP calls and response serialisation. To handle more requests at once, you start more workers (processes or threads), and each one costs memory.

FastAPI runs on Starlette, which runs on asyncio. One worker can handle thousands of requests at the same time, as long as each request spends most of its time waiting for I/O. A 500 ms call to another API doesn't block anything. The loop runs other coroutines until the response arrives. Node.js uses the same model, and Python uses it through async/await.

The catch: async only helps when your code awaits. CPU work, blocking libraries and time.sleep inside an async def route freeze the event loop, and the worker can then handle far fewer requests. The main skill is knowing the rule below.

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: an 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. In this example, the worker goes from handling thousands of requests per second to about 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. A Typical 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 returns {"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. If you send two hundred requests at the same time to this endpoint on one worker, they are all in progress at once: their waits overlap.

The connection pooling lives inside the client. Building a new AsyncClient for each request is correct here, but wasteful. In a busy app, create one shared client instead, 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 send them at the same time, not one after another. Three 200 ms calls in series take 600 ms. At the same time, they take about 200 ms.

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 (by default the others keep running, but you can pass return_exceptions=True to collect errors instead). On Python 3.11+, asyncio.TaskGroup is the modern replacement, with clearer rules for cancellation.

5. Async Databases

Every major database now has good async libraries. Here are two examples. Pick the one that fits the project.

asyncpg for PostgreSQL, a fast driver that does one job:

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, an ORM you use with await:

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 or engine is shared across the application. Open it once at startup (lifespan handles this), and inject the session into each request with a dependency.

The blocking equivalents (psycopg2, sync SQLAlchemy) work too, but only inside def routes (so FastAPI runs them in the threadpool) or through await asyncio.to_thread(...). Using a blocking driver inside an async def handler is a very common FastAPI performance bug.

6. Background Tasks: Work That Runs After the Response

Sometimes you want to return a response at once, and then do some work: log an event, send a welcome email, or 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.

Use it for short, best-effort work, where a failure should not reach the client. Don't use it for long-running jobs, work that must retry, or work you want to monitor or scale separately. 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 the same process. If the worker dies after responding but before the task finishes, the work is lost.

7. Lifespan: Startup and Shutdown the Modern Way

To load an ML model, open a DB pool or start 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, and code after it runs at shutdown. The example stores the pool on app.state directly. (If you yield a dict instead, its values are available on request.state in each request.) 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 when the module is imported (hard to test, and it is never closed properly on shutdown). With lifespan, the resource lives exactly as long as the app.

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 server does not wait for the whole response before sending. Use this for CSV exports of millions of rows, token streams from LLMs, progress updates, or data you read piece by piece from another 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 reads the file in chunks for you.

9. WebSockets: Two-Way, Long-Lived Connections

HTTP is request/response. A WebSocket is a persistent two-way connection. Once the connection is 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 a standard chat-room example. WebSockets are a good fit for chat, live dashboards, multiplayer games, and anywhere you would otherwise use 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 works 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 inside the marker. The same shape extends to Depends(...) (see fastapi-deps) and any custom validator.

11. CPU-Bound Work: Move It 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(...)):
    # Hashing 50 MB with SHA-256 is CPU-bound and would block the event loop
    digest = await asyncio.to_thread(
        lambda: hashlib.sha256(payload).hexdigest()    # hash in the worker thread, not on the loop
    )
    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+) does it in one line. For process-pool parallelism (multiple cores, no GIL limit), 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 change the route to def, and FastAPI runs it in the threadpool for you. But for anything I/O-heavy, rewrite it with httpx.AsyncClient.)

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 work, and the event loop is frozen the whole time

+ 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 are left open, and integration tests show warnings. Use lifespan for any resource that must be opened and closed: 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. There are no retries, no priorities and no monitoring. For anything that must complete, use a real queue. For small jobs like "log this" or "send an email", BackgroundTasks is fine.

6. asyncio.run inside a route

python

@app.get("/oops")
async def oops():
    asyncio.run(other_coro())           # RuntimeError because an event 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 top level of a synchronous program. Never call it inside a coroutine or 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 that one slow service doesn't make the whole request fail.

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, and the whole route should not return 500.

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 and raise 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 the whole batch fails.

Hint 2: The 3.11+ timeout context

asyncio.timeout(5.0) is the modern way to set a time limit on a block: async with asyncio.timeout(5.0): result = await gather(...). A TimeoutError is raised if the time runs out. Catch it in the route and turn it into 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 URL. 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..."}
# }

You can use the pattern in this solution for any route that builds its response from several upstream calls: dashboard endpoints, recommendation feeds, search-results pages. The four parts are:

One shared AsyncClient: all three calls reuse the same connection pool. Building a new client for each call would lose this benefit.
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 time limit on the whole block. Without it, a slow upstream can keep your route waiting longer than any client will wait.
asyncio.gather: sends the calls at the same time, returns results in order, and needs one await for the whole batch.

For very large fan-outs (50+ upstreams), add a Semaphore to limit how many requests run at once. This is the 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, and def runs on a threadpool. async def + a blocking call freezes the loop. This is the most typical 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. Use a real queue 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. For SSE, you only need to set the media type to text/event-stream.
@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 or hashing work directly in the route.
Annotated[Type, Marker(...)] is the modern syntax for Path/Query/Header/Depends metadata.

Next: FastAPI Dependency Injection: small, reusable and 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. A Typical Async Endpoint

4. Concurrent Dependent Calls with asyncio.gather

5. Async Databases

6. Background Tasks: Work That Runs After the Response

7. Lifespan: Startup and Shutdown the Modern Way

8. Streaming Responses

9. WebSockets: Two-Way, Long-Lived Connections

10. The Annotated Pattern: Modern Parameter Metadata

11. CPU-Bound Work: Move It 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