🟩 FastAPI

Dependency Injection

📖 Lesson 39 ⏱ 45 min 🧪 5 questions 💻 3 exercises

🎯 Learning Objectives

  • Understand what dependency injection is and why FastAPI uses it
  • Write simple function-based dependencies with Depends()
  • Use dependencies for query parameter parsing, pagination, and shared logic
  • Write class-based dependencies with __call__
  • Use yield dependencies for setup/teardown (database sessions, file handles)
  • Nest dependencies and share them across routers with dependencies=[]
  • Override dependencies in tests with app.dependency_overrides

1 — What is Dependency Injection?

Dependency Injection (DI) is a pattern where a function declares what it needs and the framework supplies those values automatically. You write what you need; the framework figures out how to provide it.

Why DI matters in a web framework:

  • Avoid repetition — auth checks, DB sessions, and pagination logic are written once and reused across every handler that needs them.
  • Testability — swap real dependencies for fakes in tests without touching application code.
  • Separation of concerns — route handlers stay focused on business logic; cross-cutting concerns live in dedicated dependency functions.

FastAPI's DI system is built around Depends(). It works for query parameters, path parameters, request headers, request bodies, database sessions, auth tokens — anything you need to provide to a handler.

How FastAPI resolves dependencies
Depends() is resolved at request time. FastAPI inspects each handler's signature, finds every Depends() argument, resolves them recursively (dependencies can depend on other dependencies), and injects the results. If two handlers in the same request share the same dependency, FastAPI calls it once per request and caches the result — unless you opt out with use_cache=False.

2 — Basic Depends() Usage

Any regular Python function can be a dependency. Declare it, then reference it inside a route signature wrapped in Depends(). FastAPI resolves its parameters exactly as it does for route parameters. 

from fastapi import FastAPI, Depends

app = FastAPI()

# A simple dependency — just a function
def get_query_params(q: str = "", skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

# Inject it into a route with Depends()
@app.get("/items")
def list_items(params: dict = Depends(get_query_params)):
    return params

@app.get("/users")
def list_users(params: dict = Depends(get_query_params)):
    # same dependency, reused across two routes
    return params
basic_depends.py

Visiting /items?q=hello&skip=5&limit=3 causes FastAPI to call get_query_params(q="hello", skip=5, limit=3) automatically and pass the returned dict as params.

Swagger UI shows dependency parameters
The dependency function participates in OpenAPI documentation. Its parameters (q, skip, limit) appear as query parameters in the auto-generated Swagger UI — no extra annotation needed.

3 — Typed Return Values & Nested Dependencies

Dependencies can return typed Pydantic models, and they can themselves depend on other dependencies — forming a chain that FastAPI resolves bottom-up. 

from fastapi import FastAPI, Depends, Header, HTTPException, Query, status
from pydantic import BaseModel

app = FastAPI()

class Pagination(BaseModel):
    skip: int = 0
    limit: int = 10

def pagination(skip: int = 0, limit: int = Query(default=10, le=100)) -> Pagination:
    return Pagination(skip=skip, limit=limit)

# ── Nested dependencies ──
def get_token(authorization: str = Header(...)):
    """Extract Bearer token from Authorization header."""
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer":
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid auth scheme")
    return token

def get_current_user(token: str = Depends(get_token)):
    """Validate token and return user id (simplified)."""
    if token == "secret-token":
        return {"user_id": 42, "role": "admin"}
    raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

@app.get("/profile")
def profile(user: dict = Depends(get_current_user)):
    # get_current_user → get_token → Authorization header
    # FastAPI resolves the whole chain automatically
    return {"welcome": user["user_id"], "role": user["role"]}
nested_deps.py

The dependency chain for /profile:

Step Who resolves it What it needs
1 get_token Authorization request header
2 get_current_user token string from get_token
3 profile handler user dict from get_current_user

FastAPI walks the chain bottom-up: it resolves get_token first, feeds the result to get_current_user, then feeds that result to profile. If any step raises an HTTPException, the chain short-circuits and the error response is returned immediately.

4 — Class-Based Dependencies

When a dependency needs configuration that varies between use sites, a class is cleaner than a function. Define __init__ to accept configuration and __call__ to handle each request. 

from fastapi import FastAPI, Depends, Query

app = FastAPI()

class PaginationDep:
    """Reusable pagination dependency with configurable max limit."""
    def __init__(self, max_limit: int = 100):
        self.max_limit = max_limit

    def __call__(
        self,
        skip: int = Query(default=0, ge=0),
        limit: int = Query(default=10, ge=1),
    ) -> dict:
        limit = min(limit, self.max_limit)
        return {"skip": skip, "limit": limit}

# Create instances with different configs
standard_pagination = PaginationDep(max_limit=100)
admin_pagination    = PaginationDep(max_limit=1000)

@app.get("/items")
def list_items(page: dict = Depends(standard_pagination)):
    return page

@app.get("/admin/items")
def admin_list(page: dict = Depends(admin_pagination)):
    return page
class_dep.py

standard_pagination and admin_pagination are two instances of the same class configured differently. FastAPI calls instance.__call__(skip=…, limit=…) per request, resolving skip and limit from query parameters automatically.

When to use class-based dependencies
Prefer class-based deps when you need to parameterise the dependency at definition time (e.g. different limits for different routers). Prefer function-based deps for simple, stateless logic with no configuration.

5 — yield Dependencies (Setup / Teardown)

When a dependency must clean up after itself (close a DB connection, release a lock, flush a buffer), use yield instead of return. FastAPI runs everything before the yield as setup and everything after as teardown — guaranteed even if the handler raises an exception. 

from fastapi import FastAPI, Depends
from typing import Generator

app = FastAPI()

# Simulated DB session
class DBSession:
    def __init__(self):
        print("DB: connection opened")
        self.committed = False

    def query(self, model):
        return [{"id": 1}, {"id": 2}]

    def commit(self):
        self.committed = True

    def close(self):
        print("DB: connection closed")

def get_db() -> Generator[DBSession, None, None]:
    db = DBSession()
    try:
        yield db          # ← value injected into route
    finally:
        db.close()        # ← always runs after response sent

@app.get("/items")
def list_items(db: DBSession = Depends(get_db)):
    return db.query("Item")
yield_dep.py

Lifecycle for a single request to /items:

  1. FastAPI calls get_db().
  2. Code before yield runs — DBSession() is created and the connection is opened.
  3. The yielded db object is injected into list_items.
  4. list_items executes and the response is prepared.
  5. FastAPI resumes get_db() after the yield — the finally block closes the connection.

The same pattern with SQLAlchemy async sessions looks like this:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session
async_db_dep.py
One yield per dependency
Only a single yield is allowed. Code before it = setup; code after it (ideally in a finally block) = teardown. If the handler raises an exception, FastAPI re-raises it after teardown completes, so cleanup always happens.

6 — Router-Level & App-Level Dependencies

Instead of adding Depends() to every single route signature, you can attach dependencies to an entire router or to the whole application. 

from fastapi import FastAPI, APIRouter, Depends, Header, HTTPException, status

app = FastAPI()

def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != "my-secret-key":
        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Invalid API key")

# ── Apply to a whole router ──
admin_router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(verify_api_key)],  # applied to ALL routes in this router
)

@admin_router.get("/users")
def admin_users():
    return [{"id": 1, "name": "Alice"}]

@admin_router.get("/stats")
def admin_stats():
    return {"total_users": 42}

app.include_router(admin_router)

# ── Apply to the whole app ──
app2 = FastAPI(dependencies=[Depends(verify_api_key)])  # every route requires the key
router_deps.py

Three levels of dependency application — choose the right scope:

Level How to apply Scope
Route def handler(x = Depends(dep)) One specific route only
Router APIRouter(dependencies=[Depends(dep)]) All routes in that router
Application FastAPI(dependencies=[Depends(dep)]) Every route in the app
Dependencies for side effects
When a dependency is applied at router or app level via dependencies=[], its return value is discarded — it runs purely for side effects (auth checks, rate limiting, logging). If you need the return value in the handler, declare the dep in the handler signature instead.

7 — Caching & use_cache

By default FastAPI caches each dependency's result for the duration of a single request. If expensive_op is used by two different dependencies in the same request, it is called only once and the result is reused. 

import time
from fastapi import FastAPI, Depends

app = FastAPI()

def expensive_op():
    print("running expensive operation...")
    time.sleep(0.1)    # simulate slow I/O
    return {"result": 42}

# Default: use_cache=True — called once per request even if used by multiple deps
@app.get("/cached")
def cached_route(
    a: dict = Depends(expensive_op),
    b: dict = Depends(expensive_op),   # same instance as `a` — called only once
):
    return {"a": a, "b": b}

# use_cache=False — force a fresh call every time
@app.get("/fresh")
def fresh_route(
    a: dict = Depends(expensive_op, use_cache=False),
    b: dict = Depends(expensive_op, use_cache=False),  # called twice
):
    return {"a": a, "b": b}
caching_deps.py
Overriding dependencies in tests
app.dependency_overrides is a plain dict mapping the original dependency callable to a replacement. Set it before a test, clear it after: 
from fastapi.testclient import TestClient

# Replace get_current_user with a fake that always returns a test user
def fake_current_user():
    return {"user_id": 99, "role": "tester"}

app.dependency_overrides[get_current_user] = fake_current_user

client = TestClient(app)
response = client.get("/profile", headers={"Authorization": "Bearer anything"})
assert response.json() == {"welcome": 99, "role": "tester"}

# Clean up after the test
app.dependency_overrides = {}
test_overrides.py

Caching is scoped to a single request, never across requests. Use use_cache=False when the dependency has per-call side effects that must execute every time (e.g. incrementing a per-call request counter or generating a fresh nonce).

Scenario Recommended setting
Expensive read (DB lookup, token validation) use_cache=True (default)
Side-effecting call (counter, nonce) use_cache=False
Swap dep in tests app.dependency_overrides[dep] = fake

Security Dependencies

The most common real-world use of DI in FastAPI is authentication and authorisation. FastAPI provides OAuth2PasswordBearer and HTTPBearer as ready-made security dependencies, but you can build your own token-validation chain with plain Depends(). 

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()

# OAuth2PasswordBearer extracts the Bearer token from the Authorization header
# tokenUrl is the endpoint that issues tokens (shown in /docs)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

class TokenData(BaseModel):
    user_id: int
    role: str

# Fake token store — in production use JWT (Lesson 40)
VALID_TOKENS = {
    "admin-token-abc": TokenData(user_id=1, role="admin"),
    "user-token-xyz":  TokenData(user_id=2, role="user"),
}

def get_current_user(token: str = Depends(oauth2_scheme)) -> TokenData:
    user = VALID_TOKENS.get(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

def require_admin(user: TokenData = Depends(get_current_user)) -> TokenData:
    if user.role != "admin":
        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Admin access required")
    return user

@app.get("/me")
def me(user: TokenData = Depends(get_current_user)):
    return user

@app.delete("/users/{user_id}")
def delete_user(user_id: int, admin: TokenData = Depends(require_admin)):
    return {"deleted": user_id, "by": admin.user_id}
security.py

The dependency chain is: delete_user → require_admin → get_current_user → oauth2_scheme → Authorization header. FastAPI resolves the whole chain and each step either succeeds or raises an HTTPException that aborts the request immediately.

Testing with dependency_overrides

One of the biggest benefits of DI is testability. Instead of mocking internals, you swap entire dependencies at the app level — no patching, no monkeypatching. 

# test_routes.py
import pytest
from fastapi.testclient import TestClient
from main import app, get_db, get_current_user
from models import DBSession, TokenData

# ── Fake DB ──
class FakeDB:
    def query(self, model):
        return [{"id": 99, "name": "Test Item"}]
    def close(self): pass

def override_get_db():
    yield FakeDB()

# ── Fake auth ──
def override_get_current_user():
    return TokenData(user_id=99, role="user")

# ── Apply overrides ──
app.dependency_overrides[get_db]           = override_get_db
app.dependency_overrides[get_current_user] = override_get_current_user

client = TestClient(app)

def test_list_items():
    resp = client.get("/items")
    assert resp.status_code == 200
    assert resp.json()[0]["id"] == 99

def test_me():
    resp = client.get("/me", headers={"Authorization": "Bearer anything"})
    assert resp.status_code == 200
    assert resp.json()["user_id"] == 99
test_routes.py
Clear overrides between test modules to avoid state leakage: app.dependency_overrides.clear() — or use a pytest fixture with yield to set up and tear down overrides per test. 
# conftest.py — scoped override fixture
import pytest
from fastapi.testclient import TestClient
from main import app, get_current_user
from models import TokenData

@pytest.fixture
def client_as_admin():
    app.dependency_overrides[get_current_user] = lambda: TokenData(user_id=1, role="admin")
    yield TestClient(app)
    app.dependency_overrides.clear()   # ← always clean up

def test_admin_delete(client_as_admin):
    resp = client_as_admin.delete("/users/5")
    assert resp.status_code == 200
conftest.py

Background Tasks via Dependencies

BackgroundTasks can be injected as a dependency to offload work (sending emails, writing audit logs) that should not delay the HTTP response. 

from fastapi import FastAPI, BackgroundTasks, Depends
import time

app = FastAPI()

def send_welcome_email(email: str):
    """Runs after the response is sent — never delays the caller."""
    time.sleep(2)          # simulate slow email provider
    print(f"Email sent to {email}")

def audit_log(user_id: int, action: str):
    print(f"AUDIT: user {user_id} performed {action}")

@app.post("/register")
def register(
    email: str,
    background_tasks: BackgroundTasks,
):
    # Do the fast synchronous work first
    user = {"id": 42, "email": email}

    # Queue slow tasks — run after response is returned to client
    background_tasks.add_task(send_welcome_email, email)
    background_tasks.add_task(audit_log, user["id"], "register")

    return {"user": user, "message": "Registration successful"}
background_tasks.py
BackgroundTasks runs in the same process as the web server — it is suitable for lightweight fire-and-forget tasks. For heavy workloads (video encoding, large data processing) use a proper task queue: Celery + Redis or ARQ. See Lesson 42 for a full treatment of background tasks.

Best Practices

  • Keep dependencies small and single-purpose — get_current_user validates the token; require_admin checks the role. Don't combine them into one giant function.
  • Always use yield for resources that need cleanup — database sessions, file handles, HTTP clients. Wrap the yield in try/finally so cleanup runs even on exceptions.
  • Apply auth at the router or app level for blanket protection — use APIRouter(dependencies=[Depends(require_auth)]) instead of adding the dep to every route individually.
  • Leverage dependency_overrides in tests — never patch internals; swap entire dependencies. This keeps tests fast (no real DB/network) and decoupled.
  • Use class-based deps when configuration is needed — RateLimiter(max_rpm=60) is cleaner than a closure or global variable.
  • Prefer use_cache=True (the default) — a single DB session per request is almost always what you want. Use use_cache=False only for deps with intentional per-call side effects.
  • Type the return value of dependencies — def get_db() -> Generator[Session, None, None] gives IDE autocompletion and mypy coverage in route handlers.

Exercises

Exercise 1 — Reusable Pagination & Sorting

Build a reusable dependency that handles both pagination and sorting:

  • Create a class QueryOptions with __call__(skip, limit, sort_by, order) where sort_by is constrained to Literal["id","name","price"] and order to Literal["asc","desc"].
  • Inject it into GET /products and GET /orders.
  • Apply skip/limit slicing and sort a fake in-memory list before returning.
  • Write a pytest test that uses TestClient and verifies the correct slice is returned.
💡 Hint 
from fastapi import FastAPI, Depends, Query
from typing import Literal

app = FastAPI()

PRODUCTS = [
    {"id": i, "name": f"Product {i}", "price": i * 1.5}
    for i in range(1, 21)
]

class QueryOptions:
    def __call__(
        self,
        skip:    int = Query(default=0, ge=0),
        limit:   int = Query(default=5, ge=1, le=20),
        sort_by: Literal["id", "name", "price"] = "id",
        order:   Literal["asc", "desc"] = "asc",
    ) -> dict:
        return {"skip": skip, "limit": limit, "sort_by": sort_by, "order": order}

query_opts = QueryOptions()

@app.get("/products")
def list_products(opts: dict = Depends(query_opts)):
    items = sorted(PRODUCTS, key=lambda x: x[opts["sort_by"]], reverse=opts["order"] == "desc")
    return items[opts["skip"]: opts["skip"] + opts["limit"]]

Exercise 2 — DB Session Lifecycle

Simulate a database session dependency with full lifecycle logging:

  • Create a FakeSession class with begin(), commit(), rollback(), and close() methods that print their actions.
  • Write a get_session() yield dependency: open → yield → commit on success → rollback on exception → always close.
  • Build two routes: POST /items (succeeds, should commit) and POST /bad (raises an exception after yield, should rollback).
  • Run the app and confirm the lifecycle logs appear in the correct order.
💡 Hint 
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

class FakeSession:
    def begin(self):    print("SESSION: begin")
    def commit(self):   print("SESSION: commit")
    def rollback(self): print("SESSION: rollback")
    def close(self):    print("SESSION: close")

def get_session():
    db = FakeSession()
    db.begin()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

@app.post("/items")
def create_item(db: FakeSession = Depends(get_session)):
    return {"status": "created"}

@app.post("/bad")
def bad_route(db: FakeSession = Depends(get_session)):
    raise HTTPException(500, "something went wrong")

Exercise 3 — Role-Based Access Control

Implement a full RBAC dependency chain:

  • Write get_token(authorization: str = Header(...)) that extracts a Bearer token.
  • Write get_current_user(token) that looks up the token in a dict of {token: UserData} and raises 401 if not found.
  • Write a factory require_role(role: str) that returns a dependency function checking user.role == role, raising 403 on failure.
  • Create three routes: GET /public (no auth), GET /user-area (any authenticated user), DELETE /admin/nuke (admin only).
  • Write three tests using dependency_overrides: unauthenticated access to public, user access to user-area, admin access to admin route, and confirm a regular user gets 403 on the admin route.
💡 Hint — require_role factory 
from fastapi import Depends

def require_role(role: str):
    def checker(user = Depends(get_current_user)):
        if user["role"] != role:
            raise HTTPException(status.HTTP_403_FORBIDDEN, detail=f"{role} role required")
        return user
    return checker

@app.delete("/admin/nuke", dependencies=[Depends(require_role("admin"))])
def nuke():
    return {"nuked": True}