FastAPI Dependency Injection

Dependency injection in FastAPI is the mechanism that lets you declare what a route handler needs, and FastAPI provides it automatically. It handles database connections, authentication, pagination, and any shared logic without you manually wiring things up in every route.

The basics: Depends

A dependency is any callable (function or class) that FastAPI calls before your route handler. The return value is injected as a parameter.

from fastapi import Depends, FastAPI, Query

app = FastAPI()

def pagination_params(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
):
    return {"skip": skip, "limit": limit}

@app.get("/items")
async def list_items(pagination: dict = Depends(pagination_params)):
    items = await get_items(
        skip=pagination["skip"],
        limit=pagination["limit"],
    )
    return items

When a request hits /items?skip=10&limit=50, FastAPI:

1. Calls pagination_params(skip=10, limit=50)
2. Passes the return value as pagination to list_items
3. Validates query parameters via the type hints in the dependency

Class-based dependencies

Classes with __init__ parameters work as dependencies. FastAPI injects constructor arguments from the request.

from fastapi import Depends, Query

class PaginationParams:
    def __init__(
        self,
        skip: int = Query(0, ge=0),
        limit: int = Query(20, ge=1, le=100),
    ):
        self.skip = skip
        self.limit = limit

@app.get("/users")
async def list_users(pagination: PaginationParams = Depends()):
    users = await db.users.find_many(
        skip=pagination.skip,
        take=pagination.limit,
    )
    return users

When using Depends() with no argument on a type-hinted class, FastAPI infers the dependency from the type annotation.

Sub-dependencies

Dependencies can depend on other dependencies, forming a chain.

from fastapi import Depends, Header, HTTPException

async def get_api_key(x_api_key: str = Header()):
    if not x_api_key:
        raise HTTPException(status_code=401, detail="Missing API key")
    return x_api_key

async def get_current_user(api_key: str = Depends(get_api_key)):
    user = await db.users.find_one({"api_key": api_key})
    if not user:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return user

async def get_active_user(user: dict = Depends(get_current_user)):
    if not user.get("is_active"):
        raise HTTPException(status_code=403, detail="Inactive user")
    return user

@app.get("/me")
async def read_me(user: dict = Depends(get_active_user)):
    return user

The resolution order is:

1. get_api_key extracts the header
2. get_current_user uses the API key to find the user
3. get_active_user checks the user is active
4. read_me receives the validated, active user

FastAPI resolves the full chain automatically and handles errors at any level.

Database session dependency

The most common dependency pattern is managing a database session per request.

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from fastapi import Depends

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"

engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users")
async def create_user(
    data: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    user = User(**data.model_dump())
    db.add(user)
    await db.flush()  # get the generated ID
    return user

The yield keyword makes this a generator dependency. Code before yield runs before the handler. Code after yield runs after the handler (cleanup). FastAPI handles the context manager lifecycle.

Authentication dependencies

Build a reusable auth system as dependencies.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token",
            )
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired",
        )
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate token",
        )
    
    user = await db.users.find_one({"id": user_id})
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found",
        )
    return user

def require_role(*roles: str):
    async def role_checker(user: dict = Depends(get_current_user)):
        if user["role"] not in roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Requires one of: {roles}",
            )
        return user
    return role_checker

# Usage
@app.get("/admin/users")
async def admin_list_users(
    user: dict = Depends(require_role("admin", "superadmin")),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User))
    return result.scalars().all()

@app.get("/posts")
async def list_posts(user: dict = Depends(get_current_user)):
    # Any authenticated user
    return await get_posts_for_user(user["id"])

Scoped dependencies

Use use_cache=True (the default) to call a dependency only once per request, even if multiple parts of the chain depend on it.

async def get_settings():
    print("Loading settings")  # Only prints once per request
    return await load_settings()

async def dep_a(settings: dict = Depends(get_settings)):
    return {"a": True, "settings": settings}

async def dep_b(settings: dict = Depends(get_settings)):
    return {"b": True, "settings": settings}

@app.get("/test")
async def test(
    a: dict = Depends(dep_a),
    b: dict = Depends(dep_b),
):
    # get_settings was called once; both a and b got the same instance
    return {"a": a, "b": b}

To force a fresh call each time, use Depends(get_settings, use_cache=False).

Router-level dependencies

Apply dependencies to all routes in a router.

from fastapi import APIRouter, Depends

# Every route in this router requires authentication
router = APIRouter(
    prefix="/api/v1",
    dependencies=[Depends(get_current_user)],
)

@router.get("/dashboard")
async def dashboard():
    return {"message": "Authenticated access"}

@router.get("/reports")
async def reports():
    return {"message": "Also authenticated"}

You can also apply dependencies at the app level:

app = FastAPI(dependencies=[Depends(log_request)])

Testing with dependency overrides

FastAPI lets you replace dependencies in tests without changing production code.

from fastapi.testclient import TestClient

# Override the database dependency
async def get_test_db():
    async with TestAsyncSession() as session:
        yield session

# Override the auth dependency
async def get_mock_user():
    return {
        "id": "test-user-id",
        "email": "test@example.com",
        "role": "admin",
    }

app.dependency_overrides[get_db] = get_test_db
app.dependency_overrides[get_current_user] = get_mock_user

client = TestClient(app)

def test_list_users():
    response = client.get("/admin/users")
    assert response.status_code == 200

def test_create_user():
    response = client.post("/users", json={
        "email": "new@example.com",
        "name": "New User",
    })
    assert response.status_code == 200
    assert response.json()["email"] == "new@example.com"

# Clean up overrides
app.dependency_overrides.clear()

This is one of the strongest features of FastAPI’s DI system. Production dependencies (real database, real auth) are swapped for test doubles with a single line, no mocking frameworks needed.

Practical patterns

Rate limiting dependency

from collections import defaultdict
import time

request_counts: dict[str, list[float]] = defaultdict(list)

async def rate_limit(
    user: dict = Depends(get_current_user),
    max_requests: int = 100,
    window_seconds: int = 60,
):
    now = time.time()
    user_id = user["id"]
    
    # Remove old timestamps
    request_counts[user_id] = [
        t for t in request_counts[user_id]
        if now - t < window_seconds
    ]
    
    if len(request_counts[user_id]) >= max_requests:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
        )
    
    request_counts[user_id].append(now)
    return user

Service layer dependency

class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()
    
    async def create(self, data: UserCreate) -> User:
        user = User(**data.model_dump())
        self.db.add(user)
        await self.db.flush()
        return user

async def get_user_service(
    db: AsyncSession = Depends(get_db),
) -> UserService:
    return UserService(db)

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    user = await service.get_by_id(user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

FastAPI’s dependency injection system is simple in concept but powerful in practice. Dependencies are just functions. Chains of dependencies compose naturally. Testing is trivial because you can swap any dependency. Start with simple function dependencies, and reach for classes and sub-dependency chains as your application grows.