Dependency injection is a design pattern where a function or class receives its required dependencies externally rather than creating them internally. In FastAPI, this system provides a robust way to manage shared logic, database sessions, and authentication flows.

• Maximizing code reusability by centralizing common logic
• Managing shared resources like database connections efficiently
• Enforcing security, authentication, and role-based access control

Function-Based Dependencies

You can define a dependency as an asynchronous or standard function. FastAPI handles both seamlessly, allowing async endpoints to use synchronous dependencies and vice versa.

from typing import Optional
from fastapi import APIRouter, Depends

api_router = APIRouter()

async def fetch_pagination(search: Optional[str] = None, skip: int = 0, take: int = 50):
    return {"search": search, "skip": skip, "take": take}

@api_router.get("/items/")
async def list_items(params: dict = Depends(fetch_pagination)):
    return params

@api_router.get("/users/")
def list_users(params: dict = Depends(fetch_pagination)):
    return params

Class-Based Dependencies

When a dependency requires numerous parameters, defining it as a function can become unwieldy. Using a class provides a cleaner structure and allows access to properties via the instance.

class PaginationParams:
    def __init__(self, search: Optional[str] = None, skip: int = 0, take: int = 50):
        self.search = search
        self.skip = skip
        self.take = take

@api_router.get("/products/")
async def list_products(pager=Depends(PaginationParams)):
    result = {}
    if pager.search:
        result.update({"search_term": pager.search})
    result.update({"range": f"{pager.skip}-{pager.skip + pager.take}"})
    return result

Nested Dependencies

Dependencies can themselves depend on other dependencies, forming a hierarchical graph. FastAPI automatically resolves the entire chain.

def get_keyword(keyword: Optional[str] = None):
    return keyword

def get_filtered_keyword(kw: str = Depends(get_keyword), prev_kw: Optional[str] = None):
    if not kw:
        return prev_kw
    return kw

@api_router.get("/search/")
async def search_data(result: str = Depends(get_filtered_keyword, use_cache=True)):
    # use_cache=True ensures that if multiple dependencies require the same sub-dependency,
    # the sub-dependency is only executed once per request, returning the cached value subsequently.
    return {"result": result}

Decorator-Level Dependencies

Sometimes you need to execute logic, such as validation, without injecting the return value into the path operation function. This is achieved by passing a list of dependencies to the dependencies parameter of the route decorator.

from fastapi import Header, HTTPException

async def validate_token(x_auth_token: str = Header(...)):
    if x_auth_token != "secure-token-123":
        raise HTTPException(status_code=401, detail="Invalid authentication token")

async def validate_secret(x_api_key: str = Header(...)):
    if x_api_key != "secret-key-456":
        raise HTTPException(status_code=403, detail="Invalid API key")
    return x_api_key

@api_router.get("/secure-data/", dependencies=[Depends(validate_token), Depends(validate_secret)])
async def read_secure_data():
    return [{"id": 1}, {"id": 2}]

Router and Application-Level Dependencies

To apply dependencies across an entire router or application, specify them during initialization. This ensures all routes within that scope execute the dependencies.

secured_router = APIRouter(dependencies=[Depends(validate_token), Depends(validate_secret)])

Dependencies with Yield

For setup and teardown logic, such as managing database connections, dependencies can use yield instead of return. The code before the yield handles initialization, while the code after handles cleanup. This feature requires Python 3.7 or later.

async def connect_db():
    session = "db_session_active"
    try:
        yield session
    finally:
        session = "db_session_closed"

async def init_cache():
    cache_instance = "cache_connected"
    try:
        yield cache_instance
    finally:
        cache_instance = "cache_disconnected"

async def init_logger(cache=Depends(init_cache)):
    logger_instance = "logger_created"
    try:
        yield logger_instance
    finally:
        logger_instance = f"logger_stopped_with_{cache}"