Advanced Dependency Injection Patterns in FastAPI

Why Go Beyond Basic Dependencies?

FastAPI’s Depends() system handles simple cases well — inject a database session, validate a token, parse query params. But real applications demand more. You need dependencies that share state, manage resource lifecycles, accept configuration parameters, and compose into complex chains.

This guide walks through the patterns that production FastAPI applications actually use.

Nested Dependencies

Dependencies can depend on other dependencies. FastAPI resolves the entire chain automatically and caches results within a single request.

from fastapi import Depends, FastAPI, HTTPException, status

app = FastAPI()

async def get_db_session():
    session = AsyncSession()
    try:
        yield session
    finally:
        await session.close()

async def get_current_user(session=Depends(get_db_session)):
    # session is injected automatically
    user = await session.execute(select(User).where(User.id == user_id))
    return user.scalar_one_or_none()

async def get_admin_user(user=Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    return user

@app.get("/admin/dashboard")
async def admin_dashboard(admin=Depends(get_admin_user)):
    return {"message": f"Welcome, {admin.username}"}

FastAPI resolves this bottom-up: get_db_session runs first, its result feeds into get_current_user, and that result feeds into get_admin_user. Each dependency runs only once per request even if multiple endpoints reference it.

Class-Based Dependencies

When a dependency needs configuration or internal state, wrap it in a class with a __call__ method.

from fastapi import Depends, Query

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

    def __call__(
        self,
        page: int = Query(1, ge=1),
        limit: int = Query(20, ge=1)
    ):
        limit = min(limit, self.max_limit)
        offset = (page - 1) * limit
        return {"offset": offset, "limit": limit, "page": page}

# Create instances with different configs
standard_paginator = Paginator(max_limit=100)
admin_paginator = Paginator(max_limit=500)

@app.get("/items")
async def list_items(pagination=Depends(standard_paginator)):
    return await fetch_items(
        offset=pagination["offset"],
        limit=pagination["limit"]
    )

@app.get("/admin/logs")
async def list_logs(pagination=Depends(admin_paginator)):
    return await fetch_logs(
        offset=pagination["offset"],
        limit=pagination["limit"]
    )

This pattern is powerful because each instance carries its own configuration. You can create multiple paginators, validators, or auth checkers with different rules.

Parameterized Dependencies

Sometimes you need a dependency factory — a function that returns a dependency based on parameters.

from typing import List

def require_permissions(required: List[str]):
    async def permission_checker(user=Depends(get_current_user)):
        user_perms = set(user.permissions)
        missing = set(required) - user_perms
        if missing:
            raise HTTPException(
                status_code=403,
                detail=f"Missing permissions: {', '.join(missing)}"
            )
        return user
    return permission_checker

@app.delete("/items/{item_id}")
async def delete_item(
    item_id: int,
    user=Depends(require_permissions(["items:delete"]))
):
    return {"deleted": item_id}

@app.post("/items")
async def create_item(
    user=Depends(require_permissions(["items:create"]))
):
    return {"created": True}

The outer function captures the parameters. The inner function is the actual dependency that FastAPI calls on each request. This gives you reusable, configurable guards.

Yield Dependencies for Resource Management

Use yield dependencies to manage resources that need setup and teardown — database sessions, file handles, locks, temporary directories.

from contextlib import asynccontextmanager
import aiofiles
import tempfile
import os

async def get_temp_workspace():
    workspace = tempfile.mkdtemp(prefix="fastapi_")
    try:
        yield workspace
    finally:
        # Cleanup happens after the response is sent
        for f in os.listdir(workspace):
            os.remove(os.path.join(workspace, f))
        os.rmdir(workspace)

async def get_db_transaction(session=Depends(get_db_session)):
    async with session.begin():
        yield session
    # Transaction commits automatically if no exception
    # Rolls back if an exception was raised

@app.post("/import")
async def import_data(
    workspace=Depends(get_temp_workspace),
    session=Depends(get_db_transaction)
):
    # workspace is a temp directory that gets cleaned up
    # session is inside a transaction that auto-commits
    file_path = os.path.join(workspace, "data.csv")
    # ... process import ...
    return {"status": "imported"}

The code before yield runs during setup. The code after yield runs during teardown, even if the endpoint raises an exception. This guarantees proper resource cleanup.

Shared State with Dependencies

For application-wide state like connection pools, caches, or configuration, combine yield dependencies with the app lifespan.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, Request
import redis.asyncio as redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: create shared resources
    app.state.redis = redis.Redis(host="localhost", port=6379)
    app.state.config = await load_config()
    yield
    # Shutdown: clean up
    await app.state.redis.close()

app = FastAPI(lifespan=lifespan)

async def get_redis(request: Request) -> redis.Redis:
    return request.app.state.redis

async def get_cache(redis_client=Depends(get_redis)):
    return CacheService(redis_client)

@app.get("/cached/{key}")
async def get_cached_value(key: str, cache=Depends(get_cache)):
    value = await cache.get(key)
    return {"key": key, "value": value}

The lifespan context manager creates resources once at startup and cleans them up at shutdown. Dependencies then pull these shared resources from app.state and pass them to endpoints.

Overriding Dependencies in Tests

One of the biggest advantages of dependency injection is testability. FastAPI lets you swap any dependency with app.dependency_overrides.

from fastapi.testclient import TestClient

def override_get_current_user():
    return User(id=1, username="testuser", is_admin=True)

def override_get_db_session():
    session = TestingSessionLocal()
    try:
        yield session
    finally:
        session.close()

app.dependency_overrides[get_current_user] = override_get_current_user
app.dependency_overrides[get_db_session] = override_get_db_session

client = TestClient(app)

def test_admin_dashboard():
    response = client.get("/admin/dashboard")
    assert response.status_code == 200
    assert "testuser" in response.json()["message"]

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

This approach lets you test endpoints in isolation without connecting to real databases or authentication services.

Combining Patterns: A Real-World Example

Here is a complete example that combines several patterns into a production-ready setup.

from fastapi import Depends, FastAPI, HTTPException, Request
from typing import Optional

class RateLimiter:
    def __init__(self, requests_per_minute: int):
        self.rpm = requests_per_minute

    async def __call__(
        self,
        request: Request,
        redis_client=Depends(get_redis)
    ):
        key = f"rate:{request.client.host}:{request.url.path}"
        current = await redis_client.incr(key)
        if current == 1:
            await redis_client.expire(key, 60)
        if current > self.rpm:
            raise HTTPException(429, "Rate limit exceeded")

standard_limiter = RateLimiter(requests_per_minute=60)
strict_limiter = RateLimiter(requests_per_minute=10)

def cache_response(ttl_seconds: int = 300):
    async def cacher(
        request: Request,
        cache=Depends(get_cache)
    ):
        cached = await cache.get(str(request.url))
        return cached  # None if not cached
    return cacher

@app.get("/products", dependencies=[Depends(standard_limiter)])
async def list_products(
    pagination=Depends(standard_paginator),
    cached=Depends(cache_response(ttl_seconds=60)),
    user=Depends(get_current_user)
):
    if cached:
        return cached
    products = await fetch_products(**pagination)
    return products

Key Takeaways

Advanced dependency injection in FastAPI follows a few core principles. Use nested dependencies to build chains of responsibility. Use classes with __call__ when you need configurable instances. Use factory functions when you need parameterized guards. Use yield for resource lifecycle management. Store shared state in app.state during the lifespan and pull it through dependencies.

These patterns keep your code modular, testable, and maintainable. Start with simple function dependencies and introduce classes and factories only when the complexity demands it.