Dependency Injection

FastAPI's Depends separates common logic (authentication, DB connections, pagination) into dependency functions that can be easily swapped out during testing.

---

Basic Depends

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

app = FastAPI()


# Dependency function: handle common query parameters
def common_parameters(
    skip: int = 0,
    limit: int = 10,
    q: str | None = None,
):
    return {"skip": skip, "limit": limit, "q": q}


# Concise with type hints
CommonParams = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
def list_items(commons: CommonParams):
    return {"params": commons, "items": []}


@app.get("/users/")
def list_users(commons: CommonParams):
    return {"params": commons, "users": []}


# Class-based dependency
class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.skip = (page - 1) * size
        self.limit = size
        self.page = page


@app.get("/products/")
def list_products(pagination: Annotated[Pagination, Depends()]):
    return {
        "page": pagination.page,
        "skip": pagination.skip,
        "limit": pagination.limit,
    }

---

Nested Dependencies

from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Annotated

app = FastAPI()

fake_tokens = {"alice-token": "Alice", "bob-token": "Bob"}
fake_users = {"Alice": {"id": 1, "role": "admin"}, "Bob": {"id": 2, "role": "user"}}


# Step 1: extract token
def get_token(x_token: Annotated[str, Header()]) -> str:
    if x_token not in fake_tokens:
        raise HTTPException(status_code=401, detail="Invalid token")
    return x_token


# Step 2: look up user from token (depends on step 1)
def get_current_user(token: Annotated[str, Depends(get_token)]) -> dict:
    username = fake_tokens[token]
    return fake_users[username]


# Step 3: check admin (depends on step 2)
def require_admin(user: Annotated[dict, Depends(get_current_user)]) -> dict:
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    return user


CurrentUser = Annotated[dict, Depends(get_current_user)]
AdminUser = Annotated[dict, Depends(require_admin)]


@app.get("/me")
def read_me(user: CurrentUser):
    return user


@app.get("/admin/dashboard")
def admin_dashboard(admin: AdminUser):
    return {"message": "Welcome Admin", "user": admin}

---

Resource Management — yield Dependencies

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

app = FastAPI()


class DBSession:
    def __init__(self, db_url: str):
        self.url = db_url
        self.queries: list[str] = []
        print(f"[DB] Connected: {db_url}")

    def execute(self, query: str) -> list:
        self.queries.append(query)
        return [{"id": 1, "data": query}]

    def close(self):
        print(f"[DB] Disconnected ({len(self.queries)} queries executed)")


DATABASE_URL = "postgresql://localhost/mydb"


# yield dependency: setup before yield, teardown after
def get_db() -> Generator[DBSession, None, None]:
    db = DBSession(DATABASE_URL)
    try:
        yield db
    finally:
        db.close()  # always runs (even on exception)


DB = Annotated[DBSession, Depends(get_db)]


@app.get("/users/")
def list_users(db: DB):
    return db.execute("SELECT * FROM users")


@app.get("/users/{user_id}")
def get_user(user_id: int, db: DB):
    return db.execute(f"SELECT * FROM users WHERE id={user_id}")

---

Global Dependencies

from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Annotated

app = FastAPI()


async def verify_api_key(x_api_key: Annotated[str, Header()]):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=403, detail="Invalid API Key")


# Apply dependency to all routes in a router
from fastapi import APIRouter

router = APIRouter(
    prefix="/api",
    dependencies=[Depends(verify_api_key)],
)


@router.get("/data")
def get_data():
    return {"data": "protected"}


@router.get("/more-data")
def get_more_data():
    return {"data": "also protected"}


# App-wide global dependency
app_with_global = FastAPI(
    dependencies=[Depends(verify_api_key)]
)

---

Dependency Overrides (Testing)

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from typing import Annotated

app = FastAPI()


# Real dependency
def get_db():
    return {"type": "production_db"}


DB = Annotated[dict, Depends(get_db)]


@app.get("/items/")
def list_items(db: DB):
    return {"db_type": db["type"]}


# Replace dependency in tests
def get_test_db():
    return {"type": "test_db"}


def test_list_items():
    app.dependency_overrides[get_db] = get_test_db  # swap

    client = TestClient(app)
    response = client.get("/items/")

    assert response.status_code == 200
    assert response.json()["db_type"] == "test_db"

    app.dependency_overrides.clear()  # restore

---

Summary

Concept	Description
Depends(fn)	Inject a dependency function
Nested dependencies	Dependencies that use other dependencies
yield dependency	setup/teardown pattern
Global dependencies	router(dependencies=[...])
dependency_overrides	Swap dependencies in tests
Annotated	Embed dependency in type hint

Dependency injection makes it easy to reuse authentication, DB sessions, and common parameters while keeping tests simple.