Chapter 5: FastAPI’s dependency injection system, the most distinctive feature of the framework. We cover Depends, sub-dependencies, scopes (request / app), class-based deps, security deps, overrides for testing, and the lifespan integration.
The basic shape
from fastapi import Depends
def get_settings():
return Settings()
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
return {"db": settings.database_url}
Depends(get_settings) says: “before calling info, call get_settings(); pass its result.” FastAPI builds a graph at startup; resolves at request time.
Yield-based dependencies
For setup + teardown:
async def get_db():
async with SessionLocal() as session:
yield session
# cleanup runs after handler
@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
return await db.query(User).all()
The yielded value is what the handler receives. After the handler completes (success or failure), the rest of the generator runs (the __aexit__ of the async with).
Sub-dependencies
A dependency can have its own dependencies:
def get_settings():
return Settings()
async def get_db(settings: Settings = Depends(get_settings)):
async with create_engine(settings.database_url) as engine:
async with AsyncSession(engine) as session:
yield session
FastAPI resolves the whole graph. Memo-ized within a single request.
Caching within a request
If two parts of the dependency graph need get_settings, FastAPI calls it once:
async def get_user(settings: Settings = Depends(get_settings)): ...
async def get_db(settings: Settings = Depends(get_settings)): ...
@app.get("/")
async def home(user = Depends(get_user), db = Depends(get_db)):
# get_settings ran ONCE, not twice
...
Default: use_cache=True. Override:
@app.get("/")
async def home(... , settings: Settings = Depends(get_settings, use_cache=False)):
...
Each request gets a fresh resolution; identical Depends instances share within the request.
Class-based dependencies
For grouped dependencies:
class Pagination:
def __init__(self, limit: int = 20, cursor: str | None = None):
self.limit = min(limit, 100)
self.cursor = cursor
@app.get("/posts")
async def list_posts(p: Pagination = Depends()):
return await db.list_posts(limit=p.limit, cursor=p.cursor)
Depends() (no arg) works because FastAPI uses the type. Depends(Pagination) is equivalent.
The class’s __init__ parameters become query/header/path params. Call site stays clean.
Security dependencies
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
payload = decode_jwt(token)
user = await db.get(User, payload["sub"])
if not user:
raise HTTPException(401)
return user
@app.get("/me")
async def me(user: User = Depends(get_current_user)):
return user
OAuth2PasswordBearer, HTTPBearer, APIKeyHeader, APIKeyCookie, APIKeyQuery, HTTPBasic — pre-built security deps. They auto-document in OpenAPI.
See Chapter 6 for full auth coverage.
Multi-level dependency chain
async def get_token(creds = Depends(oauth2_scheme)) -> str: return creds
async def get_current_user(token: str = Depends(get_token), db = Depends(get_db)) -> User: ...
async def get_active_user(user: User = Depends(get_current_user)) -> User:
if not user.active: raise HTTPException(403)
return user
async def get_admin_user(user: User = Depends(get_active_user)) -> User:
if not user.is_admin: raise HTTPException(403)
return user
@app.get("/admin/users")
async def admin_list(_: User = Depends(get_admin_user)):
...
Composes layers: token → user → active → admin. Each step adds one check.
Path-level dependencies
For deps that don’t return anything but do something:
async def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != EXPECTED:
raise HTTPException(401)
@app.get("/items", dependencies=[Depends(verify_api_key)])
async def items():
return [...]
dependencies=[...] runs them; their return is ignored. For “must pass this check” patterns.
Router-level dependencies
router = APIRouter(dependencies=[Depends(verify_api_key)])
Every route in the router runs verify_api_key.
app.include_router(router, dependencies=[Depends(rate_limit)])
Layered: include + router + path-level all run.
App-level dependencies
app = FastAPI(dependencies=[Depends(global_logger)])
For every request. Useful for logging / tracing setup.
Lifespan
For startup / shutdown resources (DB pools, Redis, ML models loaded once):
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# startup
app.state.db = await create_pool()
app.state.redis = await aioredis.from_url(REDIS_URL)
yield
# shutdown
await app.state.db.close()
await app.state.redis.close()
app = FastAPI(lifespan=lifespan)
Now access via Request:
async def get_db(request: Request):
return request.app.state.db
Or via Depends:
async def get_db(request: Request) -> AsyncSession:
async with request.app.state.db_factory() as session:
yield session
Settings as a dependency
from functools import lru_cache
@lru_cache
def get_settings() -> Settings:
return Settings()
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
...
lru_cache ensures Settings instantiates once.
Per-request state
async def get_correlation_id(request: Request) -> str:
return request.headers.get("x-request-id") or str(uuid.uuid4())
@app.get("/")
async def home(rid: str = Depends(get_correlation_id)):
log.info("request", request_id=rid)
Dependencies run per-request. Use them for per-request derived values.
Overrides for testing
def override_get_db():
return TestDB()
app.dependency_overrides[get_db] = override_get_db
# in tests:
client = TestClient(app)
response = client.get("/users")
# cleanup:
app.dependency_overrides = {}
Override any Depends. Non-invasive. Used heavily in tests; see Chapter 10.
Sync vs async dependencies
def sync_dep(): ...
async def async_dep(): ...
@app.get("/")
async def home(a = Depends(sync_dep), b = Depends(async_dep)):
...
Both work. Sync deps run in a threadpool; async deps run in the event loop. Mixing is fine.
For DB / HTTP / IO: prefer async (no thread overhead).
Generators (sync) and async generators
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
async def get_db_async():
async with SessionLocal() as session:
yield session
Both supported. Use whichever matches your IO library.
Cleanup ordering
async def a():
print("a setup")
yield "a"
print("a cleanup")
async def b(a_val = Depends(a)):
print("b setup")
yield "b"
print("b cleanup")
@app.get("/")
async def home(b_val = Depends(b)):
print("handler")
Output order:
a setup
b setup
handler
b cleanup
a cleanup
LIFO. Inner deps clean up first.
Errors in dependencies
If a dependency raises, the handler isn’t called; the dependency’s exception propagates:
async def auth_check(token = Header(...)):
if not valid(token):
raise HTTPException(401)
Cleanup of already-resolved deps still runs.
Class deps with inject
For services with multiple methods, inject once:
class UserService:
def __init__(self, db: AsyncSession, mailer: Mailer):
self.db = db
self.mailer = mailer
async def create(self, data): ...
async def get_user_service(
db: AsyncSession = Depends(get_db),
mailer: Mailer = Depends(get_mailer),
) -> UserService:
return UserService(db, mailer)
@app.post("/users")
async def create_user(data: UserCreate, svc: UserService = Depends(get_user_service)):
return await svc.create(data)
DI at the boundary; constructor injection inside services. Clean separation.
Performance
DI graph resolution is cheap (microseconds). Heavy work in deps adds up — that’s the dep’s fault, not the framework’s.
For per-request DB session: reuses pool connection; cheap.
For lifespan-loaded models: free per request.
Common mistakes
1. Heavy work in dependency factory
async def get_db():
return await create_engine(...).connect() # one connection per request — pool ignored
Use a pool; yield a session from it. Lifespan-managed pool; per-request session.
2. Mutable defaults across requests
def get_state(state = []): # BAD: mutable default; leaks across requests
...
Default factory or generate per-call.
3. Module-level state in deps
Deps that read module-level state (COUNTER += 1) — race conditions across worker threads.
4. Forgetting cleanup
def get_db():
return SessionLocal() # never closed
Use yield + cleanup.
5. Circular dependencies
A depends on B depends on A. Refactor.
Real-world patterns
Tenant-scoped session
async def get_tenant(request: Request) -> Tenant:
tid = request.headers.get("x-tenant-id")
return await load_tenant(tid)
async def get_tenant_db(tenant: Tenant = Depends(get_tenant)) -> AsyncSession:
async with get_session_for_tenant(tenant) as session:
yield session
See Multi-Tenancy Patterns .
Feature flag
async def feature_enabled(flag: str, user: User = Depends(get_current_user)) -> bool:
return await flags.is_enabled(flag, user.id)
# use as path-level dep
async def require_beta(enabled: bool = Depends(lambda u: feature_enabled("beta", u))):
if not enabled: raise HTTPException(403)
What’s next
Chapter 6: Authentication and Authorization.
Read this next
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .