Search Topics

Search across all FastAPI topics

GitHub

Dependency Injection

Core Concept

Dependency injection in FastAPI lets you declare what your endpoint needs, and the framework provides it. Database sessions, auth, pagination — all handled through Depends().

Every endpoint needs a database session. You copy-paste the session creation code into 30 endpoints. One day you change the connection string and miss 3 endpoints. They crash at 2 AM when the night batch job hits them.

terminal
# 2:47 AM — Alert from production

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError)
could not connect to server: Connection refused
  Is the server running on host "old-db.internal" and accepting
  TCP/IP connections on port 5432?

# Endpoint: POST /reports/generate
# This endpoint still has the OLD connection string.
# You updated 27 out of 30 endpoints. Missed 3.

Question

What if you could define your database connection once, and every endpoint that needs it just... gets it? No copy-pasting. No hunting through files. Change the connection string in one place and every endpoint picks it up automatically.

Two Approaches, One Winner

Here's the difference between copy-pasting and dependency injection. Same problem, wildly different outcomes.

The copy-paste way

Copy-paste

30 files get the same code

Change one thing

New connection string

Miss 3 endpoints

Find-and-replace isn't perfect

2 AM crash

Production down

The dependency injection way

1 function

Define it once with Depends()

Change once

Update the one function

All updated

Every endpoint uses the new version

The Problem

What you just learned

Copy-pasting setup code across endpoints creates a maintenance nightmare

Missing even one endpoint during a change can cause production crashes

Dependency injection means defining shared logic once and injecting it everywhere

Basic Dependencies

A dependency is just a callable that FastAPI runs before your endpoint. You declare what you need using Depends() and FastAPI handles the rest. That's the entire mental model.

main.py
from fastapi import Depends, FastAPI

app = FastAPI()

# Step 1: Define a function that returns what you need
async def common_parameters(
    skip: int = 0,
    limit: int = 100,
):
    return {"skip": skip, "limit": limit}

# Step 2: Declare it as a dependency — FastAPI calls it for you
@app.get("/items")
async def list_items(params: dict = Depends(common_parameters)):
    return {"params": params}

# Step 3: Reuse it anywhere — same function, zero copy-paste
@app.get("/users")
async def list_users(params: dict = Depends(common_parameters)):
    return {"params": params}

Insight

Notice how common_parameters takes query params skip and limit? FastAPI automatically extracts those from the request URL. Your dependency can use all the same parameter types as your endpoint — query params, headers, body, path params — everything.

The Annotated Pattern

Python 3.9+ introduced Annotated, which FastAPI adopted as the recommended way to declare dependencies. Instead of repeating Depends(common_parameters) everywhere, you create a reusable type alias.

main.py
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(
    skip: int = 0,
    limit: int = 100,
):
    return {"skip": skip, "limit": limit}

# Create a reusable type alias — define the injection ONCE
CommonParams = Annotated[dict, Depends(common_parameters)]

# Now it's just a type hint — clean and readable
@app.get("/items")
async def list_items(params: CommonParams):
    return {"params": params}

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

Dependency Basics

What you just learned

Depends() tells FastAPI to call a function and inject its return value

Dependencies can use query params, headers, body — anything an endpoint can

Annotated[Type, Depends()] creates reusable type aliases for cleaner code

Sub-Dependencies: Chains of Trust

Here's where it gets powerful. Dependencies can depend on other dependencies. Your endpoint needs a user? That user comes from a token. That token needs a database session. FastAPI resolves the entire chain automatically.

main.py
# Layer 1: Database session
async def get_db():
    db = SessionLocal()
    try:
        yield db            # Injected into anything that needs it
    finally:
        db.close()          # Cleanup after the request

# Layer 2: Current user (depends on Layer 1)
async def get_current_user(
    token: str = Header(),
    db: Session = Depends(get_db),  # FastAPI resolves get_db first
):
    user = db.query(User).filter_by(token=token).first()
    if not user:
        raise HTTPException(401)
    return user

# Layer 3: Your endpoint (depends on Layer 2, which depends on Layer 1)
@app.get("/me")
async def read_me(user: User = Depends(get_current_user)):
    # FastAPI resolved: get_db → get_current_user → read_me
    return user

Class-Based Dependencies

Sometimes you need a configurable dependency. Maybe your items endpoint allows up to 100 results per page, but your logs endpoint caps at 50. Same pagination logic, different limits.

dependencies.py
from fastapi import Depends, Query

class Paginator:
    def __init__(self, max_limit: int = 100):
        self.max_limit = max_limit

    # FastAPI calls this on each request
    def __call__(
        self,
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=1),
    ) -> dict:
        return {
            "skip": skip,
            "limit": min(limit, self.max_limit),  # Cap it
        }

# Configure once, inject everywhere
paginate_items = Paginator(max_limit=100)  # Items: up to 100
paginate_logs = Paginator(max_limit=50)    # Logs: up to 50

@app.get("/items")
async def list_items(pagination: dict = Depends(paginate_items)):
    return pagination

@app.get("/logs")
async def list_logs(pagination: dict = Depends(paginate_logs)):
    return pagination

Yield Dependencies: Setup + Cleanup

Use yield for dependencies that need cleanup after the response. Database sessions, file handles, temporary resources — anything you open must be closed.

main.py
async def get_db():
    db = SessionLocal()
    try:
        yield db  # <-- Injected into your endpoint
    finally:
        db.close()  # <-- Runs AFTER the response is sent
        # Even if the endpoint raised an exception!

@app.get("/items")
async def list_items(db: Session = Depends(get_db)):
    # db is ready to use — FastAPI opened it for you
    return db.query(Item).all()
    # After this returns, FastAPI runs db.close() automatically

Watch out

The code after yield always runs — even if your endpoint throws an exception. That's why you wrap it in try/finally. If you forget the finally, a failed request could leak a database connection.

Advanced Patterns

What you just learned

Dependencies can depend on other dependencies — FastAPI resolves the full chain

Class-based dependencies let you create configurable, reusable logic

yield dependencies handle setup AND cleanup — perfect for DB sessions

FastAPI caches dependency results per-request, so the same dep isn't called twice

Router-Level Dependencies

Apply a dependency to every route in a router at once. This is how you protect entire sections of your API — no need to add Depends() to each endpoint.

routers/admin.py
from fastapi import APIRouter, Depends, Header, HTTPException

async def verify_admin_token(x_admin_token: str = Header()):
    if x_admin_token != "admin-secret":
        raise HTTPException(status_code=403, detail="Not an admin")

router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(verify_admin_token)],  # Applies to ALL routes
)

@router.get("/stats")
async def admin_stats():
    # verify_admin_token runs automatically
    return {"users": 142, "active": 89}

@router.delete("/cache")
async def clear_cache():
    # verify_admin_token runs here too — zero extra code
    return {"status": "cleared"}

Go Deeper: Testing with Overrides

One of DI's biggest benefits: swap out real dependencies for fakes in tests. No monkey-patching, no mocking frameworks — just tell FastAPI "use this instead."

test_main.py
from fastapi.testclient import TestClient
from main import app, get_db

# Create a fake DB for testing
def get_test_db():
    db = TestSessionLocal()
    try:
        yield db
    finally:
        db.close()

# Swap the real DB for a test DB — one line!
app.dependency_overrides[get_db] = get_test_db

client = TestClient(app)

def test_list_items():
    response = client.get("/items")
    assert response.status_code == 200

# Clean up after tests
app.dependency_overrides.clear()

Circular Dependency Crash

You create two dependencies that depend on each other. Everything works fine... until someone actually hits the endpoint.

Broken code
main.py
async def get_service_a(
    b = Depends(get_service_b),  # A needs B
):
    return ServiceA(b)

async def get_service_b(
    a = Depends(get_service_a),  # B needs A — circular!
):
    return ServiceB(a)

@app.get("/data")
async def get_data(a = Depends(get_service_a)):
    return a.fetch()

See It In Action

When a request arrives, FastAPI resolves each dependency in order, like stations on a pipeline. Each station produces a value and passes it to the next. Hit Send Request and watch the pipeline flow — try the Admin preset to see a 3-deep chain.

Think about it...

If dependency A depends on B, and B depends on A, what happens when FastAPI tries to resolve the chain?

Hint: Think about what happens when FastAPI tries to call A, which needs B, which needs A...

Key Points

Depends()

Declare dependencies as function parameters

Composable

Dependencies can depend on other dependencies

Yield + Cleanup

Use yield for resources that need cleanup after use

Cacheable

Same dependency called twice in one request runs only once

Annotated

Use Annotated[Type, Depends()] for clean, reusable injection

Overridable

Swap dependencies in tests with app.dependency_overrides