Skip to main content

Dependency Injection

FastAPI's dependency injection system is one of its most powerful features, enabling code reuse, testing, security, and database connections in a clean, maintainable way.

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 implements DI declaratively using the Depends function.

Basic Dependencies​

Simple Function Dependency​

from fastapi import Depends, FastAPI

app = FastAPI()

def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
    return commons

Class-Based Dependencies​

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends()):
    # Shortcut: Depends() without argument when type is specified
    return commons

Dependency Chains​

Nested Dependencies​

def verify_token(x_token: str = Header(...)):
    if x_token != "secret-token":
        raise HTTPException(status_code=400, detail="Invalid token")
    return x_token

def verify_key(x_key: str = Header(...)):
    if x_key != "secret-key":
        raise HTTPException(status_code=400, detail="Invalid key")
    return x_key

def get_current_user(token: str = Depends(verify_token)):
    # This depends on verify_token
    user = decode_token(token)
    return user

@app.get("/items/")
def read_items(
    user: User = Depends(get_current_user),
    key: str = Depends(verify_key)
):
    # Both dependencies are resolved automatically
    return {"user": user, "key": key}

Multi-Level Dependencies​

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

def get_current_user(
    db: Session = Depends(get_db),
    token: str = Depends(oauth2_scheme)
):
    user = get_user_from_token(db, token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return user

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

@app.get("/users/me")
def read_user_me(current_user: User = Depends(get_current_active_user)):
    return current_user

Database Dependencies​

Database Session​

from sqlalchemy.orm import Session

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

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Connection Pooling​

class DatabaseSession:
    def __init__(self):
        self.db = SessionLocal()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.db.close()

async def get_db():
    async with DatabaseSession() as db:
        yield db

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

Security Dependencies​

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("/secure-data/")
def get_secure_data(api_key: str = Depends(verify_api_key)):
    return {"data": "This is secure"}

OAuth2 Authentication​

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

Permission Checking​

def require_permission(permission: str):
    def permission_checker(user: User = Depends(get_current_user)):
        if permission not in user.permissions:
            raise HTTPException(
                status_code=403,
                detail="Not enough permissions"
            )
        return user
    return permission_checker

@app.delete("/items/{item_id}")
def delete_item(
    item_id: int,
    user: User = Depends(require_permission("delete:items"))
):
    return {"message": "Item deleted"}

Dependency Scopes​

Request-Scoped Dependencies​

Default behavior - dependency is called once per request:

request_count = 0

def count_requests():
    global request_count
    request_count += 1
    return request_count

@app.get("/count/")
def get_count(count: int = Depends(count_requests)):
    return {"count": count}

Cached Dependencies​

Use use_cache=True to cache within a single request:

def expensive_operation():
    # This will only run once per request even if used multiple times
    return perform_expensive_calculation()

@app.get("/data/")
def get_data(
    data1: dict = Depends(expensive_operation, use_cache=True),
    data2: dict = Depends(expensive_operation, use_cache=True)
):
    # expensive_operation is called only once
    return {"data1": data1, "data2": data2}

Sub-Dependencies​

Sharing Sub-Dependencies​

async def get_user_service(db: Session = Depends(get_db)):
    return UserService(db)

async def get_item_service(db: Session = Depends(get_db)):
    return ItemService(db)

@app.get("/complex-operation/")
async def complex_operation(
    user_service: UserService = Depends(get_user_service),
    item_service: ItemService = Depends(get_item_service)
):
    # Both services share the same db session (cached)
    users = await user_service.get_users()
    items = await item_service.get_items()
    return {"users": users, "items": items}

Global Dependencies​

Application-Wide Dependencies​

async def verify_license(x_license: str = Header(...)):
    if not is_valid_license(x_license):
        raise HTTPException(status_code=403, detail="Invalid license")
    return x_license

app = FastAPI(dependencies=[Depends(verify_license)])

# All routes now require valid license header
@app.get("/items/")
def read_items():
    return {"items": []}

Router-Level Dependencies​

from fastapi import APIRouter

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

@router.get("/users/")
def read_admin_users():
    # Automatically requires admin verification
    return {"users": []}

@router.delete("/users/{user_id}")
def delete_user(user_id: int):
    # Also requires admin verification
    return {"deleted": user_id}

app.include_router(router)

Dependency Providers​

Factory Pattern​

class ServiceFactory:
    @staticmethod
    def get_email_service():
        if settings.environment == "production":
            return ProductionEmailService()
        return MockEmailService()

def get_email_service():
    return ServiceFactory.get_email_service()

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

Configuration Dependencies​

from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    admin_email: str
    items_per_page: int = 50

    class Config:
        env_file = ".env"

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

@app.get("/info/")
def get_app_info(settings: Settings = Depends(get_settings)):
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email
    }

Advanced Patterns​

Context Managers as Dependencies​

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan_context():
    # Setup
    resource = await create_resource()
    try:
        yield resource
    finally:
        # Teardown
        await cleanup_resource(resource)

async def get_resource():
    async with lifespan_context() as resource:
        yield resource

@app.get("/use-resource/")
async def use_resource(resource = Depends(get_resource)):
    return {"resource": resource}

Dependency Overrides for Testing​

# Original dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Test override
def override_get_db():
    db = TestSessionLocal()
    try:
        yield db
    finally:
        db.close()

# In tests
app.dependency_overrides[get_db] = override_get_db

Conditional Dependencies​

def get_user_or_none(token: str | None = Header(None)):
    if token:
        return get_user_from_token(token)
    return None

@app.get("/items/")
def read_items(
    user: User | None = Depends(get_user_or_none)
):
    if user:
        return get_user_items(user)
    return get_public_items()

Dependency Classes​

Callable Classes​

class RateLimiter:
    def __init__(self, requests_per_minute: int):
        self.requests_per_minute = requests_per_minute
        self.requests = {}

    def __call__(self, request: Request):
        client_ip = request.client.host
        now = time.time()

        if client_ip not in self.requests:
            self.requests[client_ip] = []

        # Clean old requests
        self.requests[client_ip] = [
            req_time for req_time in self.requests[client_ip]
            if now - req_time < 60
        ]

        if len(self.requests[client_ip]) >= self.requests_per_minute:
            raise HTTPException(status_code=429, detail="Rate limit exceeded")

        self.requests[client_ip].append(now)
        return True

rate_limiter = RateLimiter(requests_per_minute=10)

@app.get("/limited/")
def limited_endpoint(_: bool = Depends(rate_limiter)):
    return {"message": "Success"}

Parameterized Dependencies​

def has_role(role: str):
    def role_checker(user: User = Depends(get_current_user)):
        if role not in user.roles:
            raise HTTPException(
                status_code=403,
                detail=f"Role {role} required"
            )
        return user
    return role_checker

@app.get("/admin/")
def admin_only(user: User = Depends(has_role("admin"))):
    return {"message": "Admin access granted"}

@app.get("/moderator/")
def moderator_access(user: User = Depends(has_role("moderator"))):
    return {"message": "Moderator access granted"}

Async Dependencies​

Async Dependency Functions​

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

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

Mixed Sync/Async Dependencies​

# Sync dependency
def get_config():
    return load_config()

# Async dependency
async def get_user(
    config: Config = Depends(get_config),
    db: AsyncSession = Depends(get_async_db)
):
    # Can use both sync and async dependencies
    user = await db.get(User, config.default_user_id)
    return user

Best Practices​

  1. Single Responsibility: Each dependency should do one thing well
  2. Reusability: Design dependencies to be reused across endpoints
  3. Clear Naming: Use descriptive names that indicate purpose
  4. Avoid Side Effects: Dependencies should be predictable and testable
  5. Use Caching: Cache expensive operations within a request
  6. Dependency Layers: Organize dependencies in logical layers
  7. Testing: Override dependencies in tests for isolation
  8. Documentation: Document complex dependency chains
  9. Error Handling: Raise appropriate HTTP exceptions in dependencies
  10. Type Hints: Always use type hints for better IDE support

Common Use Cases​

Authentication Flow​

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    user = await authenticate_user(db, token)
    if not user:
        raise HTTPException(status_code=401)
    return user

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

Service Layer​

def get_user_service(db: Session = Depends(get_db)):
    return UserService(db)

@app.post("/users/", response_model=User)
async def create_user(
    user: UserCreate,
    service: UserService = Depends(get_user_service)
):
    return await service.create_user(user)

Background Task Management​

def get_task_queue():
    return TaskQueue()

@app.post("/process/")
async def process_data(
    data: dict,
    queue: TaskQueue = Depends(get_task_queue)
):
    task_id = await queue.add_task(process_data_task, data)
    return {"task_id": task_id}

Dependency injection in FastAPI provides a clean, testable, and maintainable way to structure your application. Master these patterns to build scalable APIs.