Dependencies and async

Reusable dependencies with Depends, yield-based resource cleanup, and the blocking-call mistake that stalls an entire async worker.

Dependencies

from typing import Annotated
from fastapi import Depends, Header, HTTPException, Query, FastAPI

app = FastAPI()

def get_token(x_token: Annotated[str | None, Header()] = None) -> str:
    if x_token != "s3cret":
        raise HTTPException(status_code=401, detail="invalid token")
    return x_token

class Pagination:
    def __init__(self,
                 limit: Annotated[int, Query(ge=1, le=100)] = 20,
                 offset: Annotated[int, Query(ge=0)] = 0):
        self.limit, self.offset = limit, offset

PageDep = Annotated[Pagination, Depends(Pagination)]

# applied to the whole router, not one handler
@app.get("/notes", dependencies=[Depends(get_token)])
def list_notes(page: PageDep) -> dict[str, int]:
    return {"limit": page.limit, "offset": page.offset}

def get_db():
    session = SessionLocal()
    try:
        yield session          # everything after yield is teardown
    finally:
        session.close()

DbDep = Annotated[Session, Depends(get_db)]

@app.post("/notes", dependencies=[Depends(get_token)])
def create_note(note: NoteIn, db: DbDep):
    db.add(to_row(note))
    db.commit()

• Dependencies are cached per request: the same callable requested twice runs once.
• Sub-dependencies compose — a dependency can itself declare Depends parameters.
• Placement options: per-endpoint (dependencies=[...]), per-router, or app-wide on FastAPI(dependencies=[...]).
• Depends works on any callable, including a class — the class signature becomes the injected schema.

async def versus def

import httpx, requests

@app.get("/slow-async")
async def slow_async():
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get("https://example.com/api")   # yields to the loop
    return r.json()

@app.get("/slow-sync")
def slow_sync():
    # plain def runs in a threadpool, so this blocking call is safe here
    return requests.get("https://example.com/api", timeout=10).json()

@app.get("/broken")
async def broken():
    # BLOCKING call inside async def: the event loop cannot serve anyone else
    return requests.get("https://example.com/api", timeout=10).json()

Testing with overrides

from fastapi.testclient import TestClient

def fake_token() -> str:
    return "test"

app.dependency_overrides[get_token] = fake_token
client = TestClient(app)

def test_list_notes():
    r = client.get("/notes?limit=5")
    assert r.status_code == 200
    assert r.json() == {"limit": 5, "offset": 0}

app.dependency_overrides.clear()   # always reset between tests

Overriding a dependency swaps the auth, database or clock for the duration of a test without touching the endpoint code. Test the real get_token separately, and the handler logic here.

FAQ

Related

Pydantic request and response models Caching and conditional requests

Last refreshed 2026-09-18.