FastAPI: Dependency injection in depth

A dependency is a function (or any callable) that FastAPI runs before your endpoint and whose result it passes in: the current user, a database session, pagination settings, a feature flag. Dependencies can depend on other dependencies, are cached per request, can run cleanup code (with yield), and can be swapped in tests with one line. Used well, they keep endpoints tiny and every cross-cutting concern in one testable place. Misused, they hide expensive work and leak resources.

1. Declaring and reusing dependencies

a function dependency, a class dependency, and Annotated aliases

from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends, FastAPI, Query
from fastapi.testclient import TestClient

def pagination(offset: Annotated[int, Query(ge=0)] = 0,
               limit: Annotated[int, Query(ge=1, le=100)] = 20) -> dict:
    return {"offset": offset, "limit": limit}            # its parameters become query parameters too

@dataclass
class Sorting:                                           # a class works too: its __init__ is the signature
    sort: str = "created"
    desc: bool = True

Page = Annotated[dict, Depends(pagination)]              # write the dependency once, reuse it everywhere
Sort = Annotated[Sorting, Depends()]                     # Depends() with no argument: "use the type itself"

app = FastAPI()

@app.get("/orders")
def list_orders(page: Page, sort: Sort):
    return {"page": page, "sort": sort}

@app.get("/customers")
def list_customers(page: Page):
    return {"page": page}

client = TestClient(app)
print(client.get("/orders?limit=5&sort=total&desc=false").json())
print(client.get("/customers?limit=500").status_code)     # the dependency's own validation applies

Output

{'page': {'offset': 0, 'limit': 5}, 'sort': {'sort': 'total', 'desc': False}}
422

2. Sub-dependencies and the per-request cache

one session per request, however many dependencies ask for it

from itertools import count
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient

session_ids = count(1)

def get_session():
    return {"session_id": next(session_ids)}          # imagine: open a DB session

Session = Annotated[dict, Depends(get_session)]

def current_user(session: Session, authorization: Annotated[str | None, Header()] = None):
    if authorization != "Bearer good-token":
        raise HTTPException(401, "not authenticated", headers={"WWW-Authenticate": "Bearer"})
    return {"user": "asha", "via_session": session["session_id"]}

User = Annotated[dict, Depends(current_user)]

app = FastAPI()

@app.get("/me")
def me(user: User, session: Session):                  # both paths lead to get_session
    return {"user": user, "endpoint_session": session["session_id"]}

@app.get("/uncached")
def uncached(a: Session, b: Annotated[dict, Depends(get_session, use_cache=False)]):
    return {"a": a["session_id"], "b": b["session_id"]}

client = TestClient(app)
auth = {"Authorization": "Bearer good-token"}
print(client.get("/me", headers=auth).json())          # the same session id twice
print(client.get("/me", headers=auth).json())          # a NEW request gets a new one
print(client.get("/uncached").json())
print(client.get("/me").status_code, client.get("/me").headers.get("www-authenticate"))

Output

{'user': {'user': 'asha', 'via_session': 1}, 'endpoint_session': 1}
{'user': {'user': 'asha', 'via_session': 2}, 'endpoint_session': 2}
{'a': 3, 'b': 4}
401 Bearer

3. yield dependencies: setup, cleanup, and scope

A dependency that yields runs its code before the yield as setup, gives the yielded value to the endpoint, and runs the code after it as cleanup. When the cleanup runs is controlled by scope (FastAPI 0.121+):

exactly when setup and cleanup run

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

events = []

def request_scoped():                    # default for yield dependencies: scope="request"
    events.append("request-dep  setup")
    try:
        yield "db-session"
    finally:
        events.append("request-dep  cleanup   (after the response was sent)")

def function_scoped():
    events.append("function-dep setup")
    try:
        yield "transaction"
    finally:
        events.append("function-dep cleanup   (before the response is sent)")

app = FastAPI()

@app.middleware("http")
async def watch(request, call_next):
    response = await call_next(request)
    events.append("middleware   has the response")
    return response

@app.get("/work")
def work(db: Annotated[str, Depends(request_scoped)],
         tx: Annotated[str, Depends(function_scoped, scope="function")]):
    events.append("endpoint     runs")
    return {"db": db, "tx": tx}

TestClient(app).get("/work")
print("\n".join(events))

Output

request-dep  setup
function-dep setup
endpoint     runs
function-dep cleanup   (before the response is sent)
middleware   has the response
request-dep  cleanup   (after the response was sent)

4. Exceptions and yield dependencies

roll back on error, and always re-raise

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.testclient import TestClient

log = []

def transaction():
    log.append("BEGIN")
    try:
        yield
        log.append("COMMIT")
    except Exception as e:
        log.append(f"ROLLBACK ({type(e).__name__})")
        raise                                       # swallowing here would turn errors into 200s / hide them

app = FastAPI()

@app.post("/transfer/{amount}", dependencies=[Depends(transaction, scope="function")])
def transfer(amount: int):
    if amount > 1000:
        raise HTTPException(409, "insufficient funds")
    return {"moved": amount}

client = TestClient(app)
print(client.post("/transfer/300").json(), log)
log.clear()
r = client.post("/transfer/5000")
print(r.status_code, r.json(), log)

Output

{'moved': 300} ['BEGIN', 'COMMIT']
409 {'detail': 'insufficient funds'} ['BEGIN', 'ROLLBACK (HTTPException)']

5. Where dependencies can be attached

endpoint, decorator, router and application level

from typing import Annotated

from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient

calls = []

def audit():                                              # side-effect only: no value needed
    calls.append("audit")

def require_api_key(x_api_key: Annotated[str | None, Header()] = None):
    if x_api_key != "k-123":
        raise HTTPException(403, "bad api key")

admin = APIRouter(prefix="/admin", dependencies=[Depends(require_api_key)])   # every admin route

@admin.get("/stats")
def stats():
    return {"orders": 42}

app = FastAPI(dependencies=[Depends(audit)])              # every route in the app
app.include_router(admin)

@app.get("/public")
def public():
    return {"ok": True}

client = TestClient(app)
print(client.get("/public").json(), client.get("/admin/stats").status_code,
      client.get("/admin/stats", headers={"X-API-Key": "k-123"}).json(), calls)

Output

{'ok': True} 403 {'orders': 42} ['audit', 'audit', 'audit']

6. Parametrised dependencies: callable instances

RoleChecker('admin'): a configurable dependency

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient

USERS = {"t-asha": {"name": "asha", "roles": {"customer"}}, "t-ravi": {"name": "ravi", "roles": {"admin"}}}

def current_user(x_token: Annotated[str, Header()]):
    if x_token not in USERS:
        raise HTTPException(401, "unknown token")
    return USERS[x_token]

class RequireRole:
    def __init__(self, *roles: str):
        self.roles = set(roles)

    def __call__(self, user: Annotated[dict, Depends(current_user)]):   # the instance is the dependency
        if not self.roles & user["roles"]:
            raise HTTPException(403, f"requires one of {sorted(self.roles)}")
        return user

app = FastAPI()

@app.post("/refunds")
def refund(user: Annotated[dict, Depends(RequireRole("admin", "support"))]):
    return {"approved_by": user["name"]}

client = TestClient(app)
print(client.post("/refunds", headers={"X-Token": "t-ravi"}).json())
r = client.post("/refunds", headers={"X-Token": "t-asha"})
print(r.status_code, r.json())

Output

{'approved_by': 'ravi'}
403 {'detail': "requires one of ['admin', 'support']"}

7. Overrides: the reason dependencies make testing easy

swap the real dependency for a fake, without touching the endpoint

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

def get_payment_gateway():
    raise RuntimeError("would call the real bank API")          # never in a test

class FakeGateway:
    def __init__(self):
        self.charges = []
    def charge(self, amount):
        self.charges.append(amount)
        return {"status": "approved", "id": f"ch_{len(self.charges)}"}

app = FastAPI()

@app.post("/pay/{amount}")
def pay(amount: int, gateway: Annotated[object, Depends(get_payment_gateway)]):
    return gateway.charge(amount)

fake = FakeGateway()
app.dependency_overrides[get_payment_gateway] = lambda: fake    # key: the ORIGINAL function
client = TestClient(app)
print(client.post("/pay/499").json(), client.post("/pay/99").json(), fake.charges)
app.dependency_overrides.clear()                                 # always reset (a pytest fixture does this)

Output

{'status': 'approved', 'id': 'ch_1'} {'status': 'approved', 'id': 'ch_2'} [499, 99]

8. Dependency pitfalls from production

Recap