FastAPI Dependency Injection Guide

1. Basic Depends

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

app = FastAPI()

# Simple dependency function
def common_params(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
) -> dict:
    return {"skip": skip, "limit": limit}

# Type alias for cleaner signatures
CommonParams = Annotated[dict, Depends(common_params)]

@app.get("/items")
async def list_items(params: CommonParams):
    return {"params": params}

@app.get("/users")
async def list_users(params: CommonParams):
    return {"params": params}

2. Class-based Dependencies

from fastapi import Depends

class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

    def to_dict(self):
        return {"page": self.page, "size": self.size, "offset": self.offset}

@app.get("/products")
async def list_products(pagination: Annotated[Pagination, Depends()]):
    # pagination.offset, pagination.size
    items = await Product.find_all(skip=pagination.offset, limit=pagination.size)
    return {"items": items, "pagination": pagination.to_dict()}

3. Yield Dependencies — DB Session

from sqlalchemy.orm import Session
from database import SessionLocal

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db          # injected into the route
    finally:
        db.close()        # always runs, even on error

DB = Annotated[Session, Depends(get_db)]

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DB):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users")
async def create_user(user_in: UserCreate, db: DB):
    user = User(**user_in.model_dump())
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

4. Sub-Dependencies

from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_token_data(token: Annotated[str, Depends(oauth2_scheme)]) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Could not validate token")

async def get_current_user(
    payload: Annotated[dict, Depends(get_token_data)],
    db: DB,
) -> User:
    user = db.query(User).filter(User.id == payload.get("sub")).first()
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

# Sub-dependency chain: route -> get_current_user -> get_token_data -> oauth2_scheme
@app.get("/me")
async def read_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

5. Router-level Dependencies

from fastapi import APIRouter

# Apply dependency to all routes in this router
router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin)],  # runs for every route
)

@router.get("/stats")
async def get_stats(db: DB):
    return await compute_stats(db)

@router.delete("/users/{user_id}")
async def delete_user(user_id: int, db: DB):
    db.query(User).filter(User.id == user_id).delete()
    db.commit()
    return {"ok": True}

6. Dependency Scopes