Dependencies Basics
FastAPI's dependency injection system lets you extract shared logic — pagination parameters, authentication checks, database sessions — into reusable functions. Any route can declare these as parameters using Depends().
By the end of this lesson you can: write dependency functions, inject them with Depends, chain dependencies, and use class-based dependencies for complex shared state.
Why Use Dependencies
Without DI, you repeat the same parameter parsing in every route:
# ❌ Repeated in every route
@router.get("/items/")
async def list_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
@router.get("/products/")
async def list_products(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
With dependencies:
from typing import Annotated
from fastapi import Query
async def pagination_params(
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 10,
) -> dict:
return {"skip": skip, "limit": limit}
from typing import Annotated
from fastapi import APIRouter, Depends
from app.dependencies.pagination import pagination_params
router = APIRouter()
PaginationDep = Annotated[dict, Depends(pagination_params)]
@router.get("/items/")
async def list_items(pagination: PaginationDep) -> dict:
return pagination
@router.get("/products/")
async def list_products(pagination: PaginationDep) -> dict:
return pagination
How Depends Works
FastAPI calls the dependency function before the handler. The return value of the dependency is injected as the parameter value.
Dependency Caching
By default, a dependency is called once per request and the result is cached — even if multiple route parameters use the same dependency:
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
async_session_factory: async_sessionmaker[AsyncSession] = ... # set at startup
async def get_db() -> AsyncSession:
"""Called once per request; result shared if Depends(get_db) appears multiple times."""
async with async_session_factory() as session:
yield session
To disable caching (call the dependency fresh each time), use Depends(my_dep, use_cache=False).
Class-Based Dependencies
For complex shared state, use a class:
from typing import Annotated
from fastapi import Depends, Query
class ItemFilter:
def __init__(
self,
q: str | None = None,
in_stock: bool | None = None,
min_price: float | None = Query(None, ge=0),
max_price: float | None = Query(None, ge=0),
):
self.q = q
self.in_stock = in_stock
self.min_price = min_price
self.max_price = max_price
FilterDep = Annotated[ItemFilter, Depends(ItemFilter)]
@router.get("/")
async def list_items(filters: FilterDep) -> dict:
return {
"q": filters.q,
"in_stock": filters.in_stock,
"price_range": [filters.min_price, filters.max_price],
}
Chaining Dependencies
Dependencies can themselves have dependencies:
from fastapi import Depends, HTTPException, Header
from typing import Annotated
async def get_token(x_api_key: Annotated[str | None, Header()] = None) -> str:
if not x_api_key:
raise HTTPException(401, "Missing API key")
return x_api_key
async def get_current_user(token: str = Depends(get_token)) -> dict:
# Validate token and look up user
if token != "valid-token":
raise HTTPException(403, "Invalid API key")
return {"user": "alice", "role": "admin"}
from app.dependencies.auth import get_current_user
from typing import Annotated
CurrentUser = Annotated[dict, Depends(get_current_user)]
@router.get("/profile")
async def get_profile(user: CurrentUser) -> dict:
return user
yield Dependencies (Cleanup)
Use yield for dependencies that need teardown (e.g., database sessions):
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db() -> AsyncSession:
async with async_session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
Code before yield is setup. Code after yield is teardown — always runs, even on exceptions.
Common Pitfalls
| Pitfall | Cause / Symptom | Fix |
|---|---|---|
| Dependency called multiple times | use_cache=False set accidentally | Remove use_cache=False unless you need fresh calls |
| Cleanup code not running | yield dependency missing try/finally | Wrap yield body in try/except/finally |
| Circular dependencies | Dep A depends on B, B depends on A | Refactor to break the cycle |
Depends on a class instance | Using Depends(MyClass()) | Use Depends(MyClass) — FastAPI instantiates it |
| Type hints not showing in docs | Using dict for dependency return | Define a Pydantic model or dataclass for the return type |
Hands-On Practice
from typing import Annotated
from fastapi import Depends, Header, HTTPException, Query
class Pagination:
def __init__(
self,
page: Annotated[int, Query(ge=1)] = 1,
per_page: Annotated[int, Query(ge=1, le=50)] = 10,
):
self.skip = (page - 1) * per_page
self.limit = per_page
self.page = page
async def require_json_accept(accept: Annotated[str, Header()] = "application/json") -> None:
if "application/json" not in accept and "*/*" not in accept:
raise HTTPException(406, "Client must accept application/json")
PageDep = Annotated[Pagination, Depends(Pagination)]
AcceptDep = Annotated[None, Depends(require_json_accept)]
from fastapi import APIRouter
from app.dependencies.common import PageDep, AcceptDep
router = APIRouter(prefix="/articles", tags=["articles"])
ARTICLES = [{"id": i, "title": f"Article {i}"} for i in range(1, 51)]
@router.get("/")
async def list_articles(pagination: PageDep, _: AcceptDep) -> dict:
items = ARTICLES[pagination.skip : pagination.skip + pagination.limit]
return {"page": pagination.page, "items": items, "total": len(ARTICLES)}
uvicorn app.routers.articles:router --reload --port 8008
curl "http://localhost:8008/articles/?page=1&per_page=5"
curl "http://localhost:8008/articles/?page=2&per_page=5"
# Test accept header check
curl "http://localhost:8008/articles/" -H "Accept: text/html" -w "\n%{http_code}"
# → 406