Python 9 min read

FastAPI Tutorial Part 5: Dependency Injection - Share Logic Across Endpoints

Master FastAPI dependency injection for clean, reusable code. Learn database sessions, authentication, pagination, and complex dependency chains.

MR

Moshiour Rahman

#fastapi #python #dependency-injection #api #tutorial #ai

Advertisement

What is Dependency Injection?

Dependency Injection (DI) is a design pattern where components receive their dependencies from external sources rather than creating them internally. FastAPI’s DI system lets you share logic, database connections, authentication, and more across endpoints.

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = DatabaseConnection()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
def get_users(db = Depends(get_db)):
    return db.query("SELECT * FROM users")

Basic Dependencies

Simple Function Dependencies

from fastapi import FastAPI, Depends

app = FastAPI()

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

@app.get("/items")
def list_items(params: dict = Depends(common_parameters)):
    return {"params": params}

@app.get("/users")
def list_users(params: dict = Depends(common_parameters)):
    return {"params": params}

Dependencies with Return Values

from datetime import datetime

def get_current_timestamp():
    return datetime.now()

def get_request_id():
    import uuid
    return str(uuid.uuid4())

@app.get("/info")
def get_info(
    timestamp: datetime = Depends(get_current_timestamp),
    request_id: str = Depends(get_request_id)
):
    return {
        "timestamp": timestamp,
        "request_id": request_id
    }

Class-Based Dependencies

Using Classes as Dependencies

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items")
def list_items(pagination: Pagination = Depends()):
    return {
        "skip": pagination.skip,
        "limit": pagination.limit
    }

Callable Classes

class ItemFilter:
    def __init__(
        self,
        q: str = None,
        min_price: float = None,
        max_price: float = None,
        category: str = None
    ):
        self.q = q
        self.min_price = min_price
        self.max_price = max_price
        self.category = category

    def apply(self, items: list) -> list:
        result = items
        if self.q:
            result = [i for i in result if self.q.lower() in i["name"].lower()]
        if self.min_price is not None:
            result = [i for i in result if i["price"] >= self.min_price]
        if self.max_price is not None:
            result = [i for i in result if i["price"] <= self.max_price]
        if self.category:
            result = [i for i in result if i["category"] == self.category]
        return result

@app.get("/items")
def list_items(filter: ItemFilter = Depends()):
    items = get_all_items()
    return filter.apply(items)

Nested Dependencies

Dependencies Can Have Dependencies

def get_db():
    return DatabaseConnection()

def get_current_user(db = Depends(get_db)):
    # Uses db dependency
    user = db.query("SELECT * FROM users WHERE id = 1")
    return user

def get_current_active_user(user = Depends(get_current_user)):
    # Uses get_current_user dependency
    if not user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return user

@app.get("/me")
def read_me(user = Depends(get_current_active_user)):
    return user

Dependency Chain Visualization

get_db()
    └── get_current_user(db)
            └── get_current_active_user(user)
                    └── Endpoint (active_user)

Database Session Management

Generator Dependencies (Yield)

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db  # Provide the session
    finally:
        db.close()  # Always cleanup

@app.get("/users")
def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

@app.post("/users")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.model_dump())
    db.add(db_user)
    db.commit()
    return db_user

Async Database Sessions

from sqlalchemy.ext.asyncio import AsyncSession

async def get_async_db():
    async with async_session() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

Path Operation Dependencies

Apply Dependencies to All Requests

async def verify_token(token: str = Header()):
    if token != "secret-token":
        raise HTTPException(status_code=401, detail="Invalid token")

async def log_request(request: Request):
    print(f"Request to {request.url}")

# Apply to specific endpoint
@app.get("/items", dependencies=[Depends(verify_token), Depends(log_request)])
def get_items():
    return {"items": []}

# Apply to all endpoints in the app
app = FastAPI(dependencies=[Depends(log_request)])

Router-Level Dependencies

from fastapi import APIRouter

router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(verify_admin_token)]
)

@router.get("/users")
def admin_get_users():
    # verify_admin_token is automatically called
    return {"users": []}

@router.delete("/users/{user_id}")
def admin_delete_user(user_id: int):
    # verify_admin_token is automatically called
    return {"deleted": user_id}

Authentication Dependencies

Token Verification

from fastapi import Header, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    if not is_valid_token(token):
        raise HTTPException(status_code=401, detail="Invalid token")
    return decode_token(token)

def get_current_user(token_data: dict = Depends(verify_token)):
    user = get_user_by_id(token_data["user_id"])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/profile")
def get_profile(user = Depends(get_current_user)):
    return user

API Key Authentication

from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

def verify_api_key(api_key: str = Depends(api_key_header)):
    if api_key not in valid_api_keys:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.get("/data", dependencies=[Depends(verify_api_key)])
def get_data():
    return {"data": "sensitive"}

Permission Checking

def require_permission(permission: str):
    def permission_checker(user = Depends(get_current_user)):
        if permission not in user.permissions:
            raise HTTPException(
                status_code=403,
                detail=f"Permission '{permission}' required"
            )
        return user
    return permission_checker

@app.delete("/users/{user_id}")
def delete_user(
    user_id: int,
    admin = Depends(require_permission("admin"))
):
    return {"deleted": user_id}

@app.get("/reports")
def get_reports(
    user = Depends(require_permission("view_reports"))
):
    return {"reports": []}

Configuration Dependencies

Settings Management

from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    debug: bool = False

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings():
    return Settings()

@app.get("/info")
def get_info(settings: Settings = Depends(get_settings)):
    return {
        "debug": settings.debug,
        "database": settings.database_url[:20] + "..."
    }

Environment-Specific Dependencies

def get_email_service(settings: Settings = Depends(get_settings)):
    if settings.debug:
        return MockEmailService()
    return RealEmailService(settings.smtp_host)

@app.post("/send-email")
def send_email(
    email: EmailSchema,
    email_service = Depends(get_email_service)
):
    email_service.send(email)
    return {"status": "sent"}

Reusable Dependencies with Parameters

Factory Pattern

def pagination(max_limit: int = 100):
    def paginate(skip: int = 0, limit: int = 10):
        return {"skip": skip, "limit": min(limit, max_limit)}
    return paginate

# Different max limits for different endpoints
@app.get("/items")
def list_items(params = Depends(pagination(max_limit=50))):
    return params

@app.get("/logs")
def list_logs(params = Depends(pagination(max_limit=1000))):
    return params

Parameterized Dependencies

def require_role(allowed_roles: list[str]):
    def role_checker(user = Depends(get_current_user)):
        if user.role not in allowed_roles:
            raise HTTPException(
                status_code=403,
                detail=f"Role must be one of: {allowed_roles}"
            )
        return user
    return role_checker

@app.get("/admin")
def admin_endpoint(user = Depends(require_role(["admin"]))):
    return {"admin": True}

@app.get("/staff")
def staff_endpoint(user = Depends(require_role(["admin", "staff"]))):
    return {"staff": True}

Dependency Caching

Cache Within Request

# Dependencies are cached per-request by default
def get_db():
    print("Creating DB connection")  # Only printed once per request
    return DatabaseConnection()

def get_user(db = Depends(get_db)):
    return db.query(User).first()

def get_user_items(db = Depends(get_db)):
    return db.query(Item).all()

@app.get("/dashboard")
def dashboard(
    user = Depends(get_user),      # Uses same db
    items = Depends(get_user_items)  # Uses same db
):
    return {"user": user, "items": items}

Disable Caching

@app.get("/random")
def get_random(
    # use_cache=False creates new instance each time
    r1: int = Depends(random_number, use_cache=False),
    r2: int = Depends(random_number, use_cache=False)
):
    return {"r1": r1, "r2": r2}  # Different values

Practical Example: Complete API with Dependencies

from fastapi import FastAPI, Depends, HTTPException, Header, Query
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
from functools import lru_cache

app = FastAPI()

# ========== Models ==========
class User(BaseModel):
    id: int
    username: str
    role: str
    is_active: bool

class Item(BaseModel):
    id: int
    name: str
    price: float
    owner_id: int

class Pagination(BaseModel):
    skip: int
    limit: int
    total: int

# ========== Settings ==========
class Settings(BaseModel):
    database_url: str = "sqlite:///./app.db"
    secret_key: str = "secret"
    debug: bool = True

@lru_cache()
def get_settings() -> Settings:
    return Settings()

# ========== Database ==========
class FakeDB:
    users = {
        1: {"id": 1, "username": "admin", "role": "admin", "is_active": True},
        2: {"id": 2, "username": "user", "role": "user", "is_active": True},
    }
    items = {
        1: {"id": 1, "name": "Widget", "price": 9.99, "owner_id": 1},
        2: {"id": 2, "name": "Gadget", "price": 19.99, "owner_id": 2},
    }

def get_db():
    db = FakeDB()
    try:
        yield db
    finally:
        pass  # Cleanup would go here

# ========== Auth Dependencies ==========
security = HTTPBearer()

def verify_token(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    token = credentials.credentials
    # In real app, decode JWT here
    if token == "admin-token":
        return {"user_id": 1}
    elif token == "user-token":
        return {"user_id": 2}
    raise HTTPException(status_code=401, detail="Invalid token")

def get_current_user(
    token_data: dict = Depends(verify_token),
    db: FakeDB = Depends(get_db)
) -> User:
    user_id = token_data["user_id"]
    if user_id not in db.users:
        raise HTTPException(status_code=404, detail="User not found")
    return User(**db.users[user_id])

def get_active_user(user: User = Depends(get_current_user)) -> User:
    if not user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return user

def require_role(roles: List[str]):
    def checker(user: User = Depends(get_active_user)) -> User:
        if user.role not in roles:
            raise HTTPException(
                status_code=403,
                detail=f"Requires role: {roles}"
            )
        return user
    return checker

# ========== Utility Dependencies ==========
class PaginationParams:
    def __init__(
        self,
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=1, le=100)
    ):
        self.skip = skip
        self.limit = limit

class ItemFilters:
    def __init__(
        self,
        min_price: Optional[float] = Query(None, ge=0),
        max_price: Optional[float] = Query(None, ge=0),
        owner_id: Optional[int] = Query(None)
    ):
        self.min_price = min_price
        self.max_price = max_price
        self.owner_id = owner_id

    def apply(self, items: dict) -> list:
        result = list(items.values())
        if self.min_price is not None:
            result = [i for i in result if i["price"] >= self.min_price]
        if self.max_price is not None:
            result = [i for i in result if i["price"] <= self.max_price]
        if self.owner_id is not None:
            result = [i for i in result if i["owner_id"] == self.owner_id]
        return result

def request_logger():
    timestamp = datetime.now()
    print(f"[{timestamp}] Request received")
    yield
    print(f"[{timestamp}] Request completed")

# ========== Endpoints ==========

# Public endpoint
@app.get("/health")
def health_check(settings: Settings = Depends(get_settings)):
    return {"status": "healthy", "debug": settings.debug}

# Authenticated endpoint
@app.get("/me", response_model=User)
def get_me(user: User = Depends(get_active_user)):
    return user

# List with pagination and filtering
@app.get("/items")
def list_items(
    pagination: PaginationParams = Depends(),
    filters: ItemFilters = Depends(),
    db: FakeDB = Depends(get_db),
    _: None = Depends(request_logger)
):
    filtered = filters.apply(db.items)
    total = len(filtered)
    paginated = filtered[pagination.skip:pagination.skip + pagination.limit]

    return {
        "items": paginated,
        "pagination": {
            "skip": pagination.skip,
            "limit": pagination.limit,
            "total": total
        }
    }

# User's own items
@app.get("/my-items")
def get_my_items(
    user: User = Depends(get_active_user),
    db: FakeDB = Depends(get_db)
):
    items = [i for i in db.items.values() if i["owner_id"] == user.id]
    return {"items": items}

# Admin only
@app.get("/admin/users")
def admin_list_users(
    admin: User = Depends(require_role(["admin"])),
    db: FakeDB = Depends(get_db)
):
    return {"users": list(db.users.values())}

@app.delete("/admin/items/{item_id}")
def admin_delete_item(
    item_id: int,
    admin: User = Depends(require_role(["admin"])),
    db: FakeDB = Depends(get_db)
):
    if item_id not in db.items:
        raise HTTPException(status_code=404, detail="Item not found")
    del db.items[item_id]
    return {"deleted": item_id}

Summary

PatternUse Case
Function dependencySimple shared logic
Class dependencyComplex logic with state
Yield dependencyResource cleanup (DB sessions)
Nested dependenciesBuild dependency chains
Router dependenciesApply to all routes in router
Parameterized dependenciesReusable with different configs
Dependency TypeExample
Database sessionget_db() with yield
Authenticationget_current_user()
PaginationPaginationParams class
Settingsget_settings() with cache
LoggingGenerator with yield
PermissionsFactory function pattern

Next Steps

In Part 6, we’ll explore Database Integration with SQLAlchemy - connecting FastAPI to databases, defining models, and performing queries.

Series Navigation:

  • Part 1: Introduction and Setup
  • Part 2: Path and Query Parameters
  • Part 3: Request Bodies and Pydantic
  • Part 4: Response Models and Status Codes
  • Part 5: Dependency Injection (You are here)
  • Part 6: Database Integration

Advertisement

MR

Moshiour Rahman

Software Architect & AI Engineer

Share:
MR

Moshiour Rahman

Software Architect & AI Engineer

Enterprise software architect with deep expertise in financial systems, distributed architecture, and AI-powered applications. Building large-scale systems at Fortune 500 companies. Specializing in LLM orchestration, multi-agent systems, and cloud-native solutions. I share battle-tested patterns from real enterprise projects.

Related Articles

Python

FastAPI Tutorial Part 20: Building a Production-Ready API

Build a complete production-ready FastAPI application. Combine all concepts into a real-world e-commerce API with authentication, database, and deployment.

Python

FastAPI Tutorial Part 8: Error Handling and Exceptions

Master error handling in FastAPI. Learn custom exceptions, global handlers, validation errors, and building user-friendly error responses.

Python

FastAPI Tutorial Part 1: Introduction and Setup - Build Modern Python APIs

Start your FastAPI journey with this comprehensive guide. Learn installation, create your first API, understand async Python, and explore automatic documentation.

Comments

Comments are powered by GitHub Discussions.

Configure Giscus at giscus.app to enable comments.