3. Dependencies & middleware¶
Intermediate · 10 min read
Dependencies are functions FastAPI runs before your endpoint and passes in as arguments — settings, the current user, a database session, an LLM client. Middleware wraps every request — logging, timing, CORS.
3.1 Depends — a shared input¶
from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
client = TestClient(app)
def pagination(page: int = 1, size: int = 10) -> dict:
return {"offset": (page - 1) * size, "limit": size}
@app.get("/documents")
def list_documents(p: Annotated[dict, Depends(pagination)]):
return p
@app.get("/chunks")
def list_chunks(p: Annotated[dict, Depends(pagination)]):
return p
print(client.get("/documents?page=3&size=20").json())
print(client.get("/chunks").json())
The dependency's own parameters (page, size) become query parameters of every endpoint that uses it.
3.2 Settings as a dependency¶
from functools import lru_cache
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
llm_model: str = "gpt-4o-mini"
service_api_key: str = "dev-key"
@lru_cache
def get_settings() -> Settings:
return Settings()
SettingsDep = Annotated[Settings, Depends(get_settings)]
@app.get("/config")
def config(settings: SettingsDep):
return {"model": settings.llm_model}
print(client.get("/config").json())
See Settings & secrets for .env files and SecretStr.
3.3 API-key authentication¶
A dependency can refuse the request. Every endpoint that declares it is protected:
from fastapi import HTTPException, Security
from fastapi.security import APIKeyHeader
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def require_api_key(settings: SettingsDep, key: str | None = Security(api_key_header)) -> str:
if key != settings.service_api_key:
raise HTTPException(status_code=401, detail="Invalid or missing API key")
return key
@app.post("/chat", dependencies=[Depends(require_api_key)])
def chat():
return {"reply": "Hello from a protected endpoint"}
print(client.post("/chat").status_code)
print(client.post("/chat", headers={"X-API-Key": "dev-key"}).json())
To protect a whole router: APIRouter(dependencies=[Depends(require_api_key)]).
Per-user keys and rate limits
Return a user object from the dependency instead of the key (look the key up in a DB). Then the endpoint knows who is calling — needed for per-user rate limits and token budgets, which matter a lot when every request costs money.
3.4 Dependencies with cleanup — yield¶
Code after yield runs after the response is sent — the place to close a DB session:
log: list[str] = []
def get_db():
log.append("open db")
try:
yield {"conn": "fake-connection"}
finally:
log.append("close db")
@app.get("/history/{session_id}")
def history(session_id: str, db: Annotated[dict, Depends(get_db)]):
log.append(f"query {session_id}")
return {"session": session_id}
client.get("/history/abc")
print(log)
3.5 Lifespan — load expensive things once¶
Embedding models, vector-store clients and HTTP clients should be created once at startup, not per request:
from contextlib import asynccontextmanager
from fastapi import Request
@asynccontextmanager
async def lifespan(app: FastAPI):
print("startup: loading embedding model + vector DB client")
app.state.embedder = lambda text: [len(text), text.count(" ")] # stand-in for a real model
yield
print("shutdown: closing clients")
app2 = FastAPI(lifespan=lifespan)
@app2.post("/embed")
def embed(text: str, request: Request):
return {"vector": request.app.state.embedder(text)}
with TestClient(app2) as c: # `with` runs startup and shutdown
print(c.post("/embed", params={"text": "hello world"}).json())
startup: loading embedding model + vector DB client
{'vector': [11, 1]}
shutdown: closing clients
3.6 Middleware — wrap every request¶
Add a request ID and timing header to every response (useful for tracing slow LLM calls). Middleware must be added when you create the app, before it serves any request — so this example starts a fresh app:
import time, uuid
web = FastAPI()
@web.get("/ping")
def ping():
return {"pong": True}
@web.middleware("http")
async def timing(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
response.headers["X-Request-ID"] = uuid.uuid4().hex[:8]
response.headers["X-Process-Time-ms"] = f"{(time.perf_counter() - start) * 1000:.1f}"
return response
web_client = TestClient(web)
r = web_client.get("/ping")
print(sorted(h for h in r.headers if h.startswith("x-")))
3.7 CORS — let a browser frontend call you¶
A React or Next.js app on another domain is blocked by the browser unless you allow it:
from fastapi.middleware.cors import CORSMiddleware
api = FastAPI()
api.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://www.aiwithpriyabrat.com"],
allow_methods=["GET", "POST"],
allow_headers=["*"],
)
@api.get("/ping")
def ping_api():
return {"pong": True}
r = TestClient(api).options("/ping", headers={"Origin": "http://localhost:3000",
"Access-Control-Request-Method": "GET"})
print(r.status_code, r.headers["access-control-allow-origin"])
Don't allow every origin (*) in production
List your real frontend domains. CORS only affects browsers — it is not authentication; you still need the API key or login check.
3.8 Overriding dependencies in tests¶
The big win of Depends: tests can swap the real thing for a fake.
app.dependency_overrides[get_settings] = lambda: Settings(llm_model="fake-model", service_api_key="t")
print(client.get("/config").json())
app.dependency_overrides.clear()
Practice¶
- Write a
get_current_userdependency that maps API keys to user names from a dict, and return the user from/me. - Add middleware that logs method, path, status code and time for every request.
Next: Async & streaming — stream LLM tokens to the user.