fastapi-dependency-injection

name	fastapi-dependency-injection
user-invocable	false
description	Master FastAPI dependency injection for building modular, testable APIs. Use when creating reusable dependencies and services.
allowed-tools	["Bash","Read"]

FastAPI Dependency Injection

Master FastAPI's dependency injection system for building modular, testable APIs with reusable dependencies.

Basic Dependencies

Simple dependency injection patterns in FastAPI.

from fastapi import Depends, FastAPI

app = FastAPI()

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

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

# Simple function dependency
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {'q': q, 'skip': skip, 'limit': limit}

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

# Async dependencies
async def get_async_db():
    db = await AsyncDatabase.connect()
    try:
        yield db
    finally:
        await db.disconnect()

@app.get('/async-users')
async def get_async_users(db = Depends(get_async_db)):
    return await db.fetch_all('SELECT * FROM users')

Dependency Scopes

Understand dependency lifecycle and caching.

from fastapi import Depends

# Request-scoped dependency (default, cached per request)
def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    return user

# Multiple uses in same request reuse the same instance
@app.get('/profile')
async def get_profile(
    user1 = Depends(get_current_user),
    user2 = Depends(get_current_user)  # Same instance as user1
):
    assert user1 is user2  # True
    return user1

# No caching (use_cache=False)
def get_fresh_data(use_cache: bool = False):
    return fetch_from_api()

@app.get('/data')
async def get_data(data = Depends(get_fresh_data, use_cache=False)):
    return data  # Fetches fresh data each time

# Singleton pattern (application scope)
class Settings:
    def __init__(self):
        self.app_name = 
        .admin_email = 

settings = Settings()  

 ():
     settings


  ():
     settings

Sub-Dependencies

Build complex dependency chains.

from fastapi import Depends, HTTPException, status

# Sub-dependency chain
def get_db():
    db = Database()
    try:
        yield db
    finally:
        db.close()

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db = Depends(get_db)
):
    user = db.query_one('SELECT * FROM users WHERE token = ?', token)
    if not user:
        raise HTTPException(status_code=401, detail='Invalid token')
    return user

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

def get_admin_user(
    current_user = Depends(get_current_active_user)
):
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail='Not authorized')
    return current_user

# Use in endpoint
@app.delete('/users/{user_id}')
async def ():
     db.execute(, user_id)
     {: }

Class-Based Dependencies

Use classes for stateful dependencies.

from fastapi import Depends

class Database:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None

    async def connect(self):
        self.connection = await create_connection(self.connection_string)
        return self

    async def disconnect(self):
        if self.connection:
            await self.connection.close()

    async def fetch_all(self, query: str):
        return await self.connection.fetch_all(query)

# Callable class (using __call__)
class DatabaseDependency:
    def __init__(self):
        self.db = None

    async def __call__(self):
        if  .db:
            .db = Database()
             .db.connect()
         .db

db_dependency = DatabaseDependency()


  ():
      db.fetch_all()


 :
     ():
        .skip = skip
        .limit = limit


  ():
     {: pagination.skip, : pagination.limit}


 :
     ():
        .api_key = api_key
        .timeout = timeout
        .max_retries = max_retries
        .client = 

      ():
          .client:
            .client =  create_http_client(
                api_key=.api_key,
                timeout=.timeout
            )
         .client


 ():
     ServiceClient(api_key=api_key, timeout=)

service = create_service_dependency()


  ():
      client.fetch()

Generator Dependencies for Cleanup

Use generator functions to ensure proper resource cleanup.

from contextlib import asynccontextmanager
from fastapi import Depends
import httpx

# Database connection with cleanup
async def get_db_connection():
    connection = await Database.connect('postgresql://localhost/db')
    try:
        yield connection
    finally:
        await connection.close()
        print('Database connection closed')

# HTTP client with cleanup
async def get_http_client():
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client
    # Client automatically closed after yield

@app.get('/external-api')
async def call_external_api(client = Depends(get_http_client)):
    response = await client.get('https://api.example.com/data')
    return response.json()

# File handle with cleanup
async def get_file_writer(filename: str):
    file = await aiofiles.open(filename, mode='a')
    try:
         file
    :
         file.close()


  ():
    db =  Database.connect()
    cache =  Redis.connect()
    logger = Logger()

    :
         {: db, : cache, : logger}
    :
         cache.close()
         db.close()
        logger.shutdown()


  ():
    db = resources[]
    cache = resources[]
    logger = resources[]

    logger.info()
    result =  db.save(data)
     cache.invalidate()
     result



  ():
      db.begin()  transaction:
        :
             transaction
             transaction.commit()
         Exception:
             transaction.rollback()
            

  ():
      transaction_context(db)  txn:
         txn


  ():
     txn.execute(
        ,
        amount, from_account
    )
     txn.execute(
        ,
        amount, to_account
    )
     {: }

Dependency Chains and Sub-Dependencies

Build complex dependency hierarchies.

from fastapi import Depends, HTTPException
from typing import Optional

# Level 1: Configuration
def get_config():
    return {
        'database_url': 'postgresql://localhost/db',
        'redis_url': 'redis://localhost',
        'secret_key': 'super-secret'
    }

# Level 2: Infrastructure (depends on config)
def get_db(config: dict = Depends(get_config)):
    db = Database(config['database_url'])
    try:
        yield db
    finally:
        db.close()

def get_cache(config: dict = Depends(get_config)):
    cache = Redis(config['redis_url'])
    try:
        yield cache
    finally:
        cache.close()

# Level 3: Authentication (depends on infrastructure)
def get_token_from_header(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail='Missing token')
    scheme, token = authorization.split()
    if scheme.lower() != 'bearer':
         HTTPException(status_code=, detail=)
     token

 ():
    :
        payload = jwt.decode(
            token,
            config[],
            algorithms=[]
        )
         payload
     JWTError:
         HTTPException(status_code=, detail=)

 ():
    user_id = payload.get()
    user = db.query_one(, user_id)
      user:
         HTTPException(status_code=, detail=)
     user


 ():
      user.is_active:
         HTTPException(status_code=, detail=)
     user

 ():
     ():
         permission   user.permissions:
             HTTPException(
                status_code=,
                detail=
            )
         user
     permission_checker


 ():
     UserService(db=db, cache=cache, user=current_user)



  ():
    result =  service.deactivate_user(user_id)
     result

OAuth2 Dependencies

Implement OAuth2 authentication with dependencies.

from fastapi import Depends, HTTPException, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    HTTPBearer,
    HTTPAuthorizationCredentials
)
from datetime import datetime, timedelta
from jose import JWTError, jwt

# OAuth2 with password flow
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

SECRET_KEY = 'your-secret-key-here'
ALGORITHM = 'HS256'
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({'exp': expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user_from_token(
    token: str = Depends(oauth2_scheme),
    db = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail='Could not validate credentials',
        headers={'WWW-Authenticate': 'Bearer'},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get('sub')
        if user_id is None:
            raise credentials_exception
     JWTError:
         credentials_exception

    user =  db.get_user_by_id(user_id)
     user  :
         credentials_exception
     user


bearer_scheme = HTTPBearer()

  ():
    token = credentials.credentials
    :
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
         payload
     JWTError:
         HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=
        )


 fastapi.security  OAuth2PasswordBearer, SecurityScopes

oauth2_scheme_scopes = OAuth2PasswordBearer(
    tokenUrl=,
    scopes={
        : ,
        : ,
        : 
    }
)

  ():
     security_scopes.scopes:
        authenticate_value = 
    :
        authenticate_value = 

    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=,
        headers={: authenticate_value},
    )

    :
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id:  = payload.get()
         user_id  :
             credentials_exception
        token_scopes = payload.get(, [])
     JWTError:
         credentials_exception

    user =  db.get_user_by_id(user_id)
     user  :
         credentials_exception

     scope  security_scopes.scopes:
         scope   token_scopes:
             HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=,
                headers={: authenticate_value},
            )
     user



  ():
     current_user


  ():
    updated_user =  db.update_user(current_user., user_update)
     updated_user

Database Dependencies

Manage database connections with dependencies.

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# SQLAlchemy setup
DATABASE_URL = 'postgresql+asyncpg://user:pass@localhost/db'
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# Dependency
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# Usage
from sqlalchemy import select

@app.get('/users/{user_id}')
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail='User not found')
    return user



  ():
    user = User(**user_data.())
    db.add(user)
     db.flush()  
     user

Authentication Dependencies

Implement authentication patterns.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')

SECRET_KEY = 'your-secret-key'
ALGORITHM = 'HS256'

# Token verification
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail='Could not validate credentials',
        headers={'WWW-Authenticate': 'Bearer'},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get('sub')
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.fetch_one('SELECT * FROM users WHERE email = ?', email)
    if user is None:
        raise credentials_exception
    return user

# Active user check
async def ():
      current_user.is_active:
         HTTPException(status_code=, detail=)
     current_user


 ():
      ():
         current_user.role != required_role:
             HTTPException(
                status_code=,
                detail=
            )
         current_user
     role_checker


  ():
     {: }

Caching Dependencies

Implement caching patterns with dependencies.

from fastapi import Depends
from functools import lru_cache
import redis.asyncio as redis

# In-memory cache
@lru_cache()
def get_settings():
    return Settings()

@app.get('/config')
async def get_config(settings: Settings = Depends(get_settings)):
    return settings

# Redis cache dependency
class RedisCache:
    def __init__(self):
        self.redis = None

    async def get_connection(self):
        if not self.redis:
            self.redis = await redis.from_url('redis://localhost')
        return self.redis

    async def get(self, key: str):
        conn = await self.get_connection()
        value = await conn.get(key)
        return value.decode() if value  

      ():
        conn =  .get_connection()
         conn.(key, value, ex=expire)

cache = RedisCache()

  ():
     cache


  ():
    value =  cache.get(key)
     value:
         {: value, : }

    
    value = fetch_expensive_data(key)
     cache.(key, value)
     {: value, : }

Background Task Dependencies

Use dependencies with background tasks.

from fastapi import BackgroundTasks, Depends
import asyncio

async def send_email(email: str, message: str):
    # Simulate sending email
    await asyncio.sleep(2)
    print(f'Email sent to {email}: {message}')

def get_email_service():
    # Could return configured email service
    return send_email

@app.post('/users')
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
    email_service = Depends(get_email_service),
    db = Depends(get_db)
):
    user_obj = await db.create_user(user)
    background_tasks.add_task(
        email_service,
        user.email,
        'Welcome to our service!'
    )
    return user_obj

# Complex background task with dependencies
class EmailService:
    def __init__(self, db = Depends(get_db)):
        self.db = db

    async def send_welcome_email(self, user_id: int):
        user = await self.db.get_user(user_id)
         send_email(user.email, )


  ():
    user_obj =  db.create_user(user)
    email_service = EmailService(db)
    background_tasks.add_task(
        email_service.send_welcome_email,
        user_obj.
    )
     user_obj


  ():
    :
        
         process_file(file_path)
    :
        
        os.remove(file_path)


  ():
    file_path = 
     (file_path, )  f:
        f.write( file.read())

    background_tasks.add_task(process_file_with_cleanup, file_path)
     {: }

WebSocket Dependencies

Use dependencies with WebSocket connections.

from fastapi import WebSocket, Depends, WebSocketDisconnect
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

def get_connection_manager():
    return manager

# WebSocket with authentication
async def get_ws_current_user(
    websocket: WebSocket,
    token: str = Query(...),
    db = Depends(get_db)
):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get('sub')
        user = await db.get_user_by_id(user_id)
          user:
             websocket.close(code=)
             HTTPException(status_code=, detail=)
         user
     JWTError:
         websocket.close(code=)
         HTTPException(status_code=, detail=)


  ():
     manager.connect(websocket)
    :
         :
            data =  websocket.receive_text()
            message = 
             manager.broadcast(message)
     WebSocketDisconnect:
        manager.disconnect(websocket)
         manager.broadcast()



  ():
     websocket.accept()

    
    messages =  db.get_room_messages(room_id)
     websocket.send_json(messages)

    :
         :
            data =  websocket.receive_text()
            
            message =  db.create_message(
                room_id=room_id,
                user_id=user.,
                content=data
            )
             websocket.send_json(message)
     WebSocketDisconnect:
        

Custom Dependency Providers

Create custom dependency injection patterns.

from typing import Callable, Type, TypeVar, Generic
from fastapi import Depends

T = TypeVar('T')

# Dependency factory
class DependencyFactory(Generic[T]):
    def __init__(self, dependency_class: Type[T], **kwargs):
        self.dependency_class = dependency_class
        self.kwargs = kwargs

    def __call__(self) -> T:
        return self.dependency_class(**self.kwargs)

# Service locator pattern
class ServiceLocator:
    def __init__(self):
        self._services = {}

    def register(self, name: str, service):
        self._services[name] = service

    def get(self, name: str):
        return self._services.get(name)

    def create_dependency(self, name: str):
        def get_service():
            service = self.get(name)
             service  :
                 ValueError()
             service
         get_service


locator = ServiceLocator()
locator.register(, Database())
locator.register(, Redis())


get_db_from_locator = locator.create_dependency()


  ():
      db.fetch_all()


 :
     ():
        .request = request
        .context = {}

     ():
        .context[key] = value

     ():
         .context.get(key)

  ():
    context = ContextualDependency(request)
    context.(, (uuid.uuid4()))
    context.(, datetime.utcnow())
     context


  ():
     {
        : context.get(),
        : context.get()
    }


 :
     ():
        .factory = factory
        ._instance = 

     ():
         ._instance  :
            ._instance = .factory()
         ._instance

 ():
    ()
     ExpensiveService()

expensive_service = LazyDependency(create_expensive_service)


  ():
     service.do_work()

Scoped Dependencies Per Request

Manage request-scoped state and lifecycle.

from contextvars import ContextVar
from fastapi import Depends, Request

# Request-scoped storage using context variables
request_id_var: ContextVar[str] = ContextVar('request_id')

async def set_request_id(request: Request):
    request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()))
    request_id_var.set(request_id)
    return request_id

async def get_request_id():
    return request_id_var.get()

# Request-scoped logger
class RequestLogger:
    def __init__(self, request_id: str = Depends(get_request_id)):
        self.request_id = request_id

    def info(self, message: str):
        print(f'[{self.request_id}] INFO: {message}')

    def error(self, message: str):
        print(f'[{self.request_id}] ERROR: {message}')

@app.get('/scoped-logging')
async def ():
    logger.info()
    
    logger.info()
     {: request_id}


 :
     ():
        .db = db
        .repositories = {}

     ():
         model_class   .repositories:
            .repositories[model_class] = Repository(.db, model_class)
         .repositories[model_class]

      ():
         .db.commit()

      ():
         .db.rollback()


  ():
    :
        user_repo = uow.get_repository(User)
        order_repo = uow.get_repository(Order)

        user =  user_repo.create(data.user)
        order =  order_repo.create(data.order, user_id=user.)

         uow.commit()
         {: user, : order}
     Exception:
         uow.rollback()
        

Global Dependencies with app.dependency_overrides

Use global dependency management and overrides.

from fastapi import FastAPI, Depends

app = FastAPI()

# Original dependencies
def get_production_db():
    return ProductionDatabase('postgresql://prod/db')

def get_production_cache():
    return RedisCache('redis://prod')

# Default app setup
@app.get('/data')
async def get_data(
    db = Depends(get_production_db),
    cache = Depends(get_production_cache)
):
    cached = await cache.get('data')
    if cached:
        return cached

    data = await db.fetch_all('SELECT * FROM data')
    await cache.set('data', data)
    return data

# Override for testing environment
if os.getenv('ENVIRONMENT') == 'test':
    def get_test_db():
        return TestDatabase(':memory:')

    def get_test_cache():
        return InMemoryCache()

    app.dependency_overrides[get_production_db] = get_test_db
    app.dependency_overrides[get_production_cache] = get_test_cache

# Override for development
if os.getenv('ENVIRONMENT') == 'development':
     ():
         DevDatabase()

    app.dependency_overrides[get_production_db] = get_dev_db


  ():
    tenant_id = request.headers.get()
     tenant_id:
         TenantDatabase(tenant_id)
     get_production_db()


 ():
     config.get():
        app.dependency_overrides[get_production_db] = : MockDB()

     config.get():
        app.dependency_overrides[get_production_cache] = : LocalCache()


 ():
    app.dependency_overrides = {}

Testing with Dependencies

Override dependencies for testing.

from fastapi.testclient import TestClient

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

app.dependency_overrides = {}

# Test with mock database
def test_get_users():
    def override_get_db():
        return MockDB()

    app.dependency_overrides[get_db] = override_get_db

    client = TestClient(app)
    response = client.get('/users')

    assert response.status_code == 200

    # Cleanup
    app.dependency_overrides = {}

# Pytest fixture for dependency override
import pytest

@pytest.fixture
def client():
    def override_get_db():
        return MockDB()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    yield client
    app.dependency_overrides = {}

def test_with_fixture(client):
    response = client.get('/users')
    assert response.status_code == 200

Global Dependencies

Apply dependencies to all routes.

from fastapi import FastAPI, Depends

# Logging dependency
async def log_request(request: Request):
    print(f'{request.method} {request.url}')

# Rate limiting
async def rate_limit(request: Request):
    client_ip = request.client.host
    # Check rate limit
    if is_rate_limited(client_ip):
        raise HTTPException(status_code=429, detail='Too many requests')

# Apply globally
app = FastAPI(dependencies=[Depends(log_request), Depends(rate_limit)])

# Apply to router
router = APIRouter(dependencies=[Depends(get_current_user)])

@router.get('/protected-resource')
async def protected_route():
    return {'message': 'This requires authentication'}

app.include_router(router)

When to Use This Skill

Use fastapi-dependency-injection when building modern, production-ready applications that require advanced patterns, best practices, and optimal performance.

FastAPI DI Best Practices

1. Use yield for cleanup - Always use yield for resources that need cleanup like database connections, file handles, and network connections to ensure proper resource management

2. Leverage caching - Use dependency caching (enabled by default) to avoid redundant work within a request; multiple uses of the same dependency in one request share the same instance

3. Chain dependencies - Build complex dependencies from simpler sub-dependencies to create composable, testable, and maintainable code structures

4. Class-based for state - Use classes for dependencies that maintain state or configuration, leveraging __call__ method for callable instances

5. Type hints everywhere - Always add type hints for better editor support, automatic validation, and improved documentation generation

6. Override for testing - Use dependency_overrides to inject mocks during testing without modifying production code

7. Global for cross-cutting - Apply common dependencies (logging, auth, rate limiting) globally or at router level to avoid repetition

8. Async when possible - Use async dependencies for I/O operations to maximize performance and concurrency benefits

9. Separate concerns - Keep authentication, authorization, and business logic in separate dependencies for better testability and reusability

10. Document dependencies - Add docstrings to explain complex dependency chains, especially when building multi-level hierarchies

11. Use Security utilities - Leverage FastAPI's security utilities like OAuth2PasswordBearer and HTTPBearer for authentication patterns

12. Validate early - Place validation dependencies early in the chain to fail fast and provide clear error messages

13. Keep dependencies pure - Dependencies should have minimal side effects; use background tasks for non-critical operations

14. Use context managers - Wrap dependencies in context managers when dealing with transactions or resource pools

15. Dependency composition - Compose larger dependencies from smaller, focused ones rather than creating monolithic dependencies

FastAPI DI Common Pitfalls

1. Forgetting yield - Not using yield means resources won't be cleaned up properly, leading to connection leaks and resource exhaustion

2. Circular dependencies - Creating dependency cycles causes infinite loops and stack overflow errors; design dependencies in a directed acyclic graph

3. Not using Depends() - Forgetting Depends() wrapper means function is called directly instead of being injected, breaking the dependency resolution

4. Overusing use_cache=False - Disabling cache unnecessarily hurts performance by creating multiple instances of the same dependency per request

5. Heavy dependencies - Putting too much logic in dependencies instead of services makes them hard to test and violates single responsibility

6. Not testing overrides - Forgetting to test with dependency_overrides means tests may use production resources instead of mocks

7. Mixing sync and async - Incorrectly mixing synchronous and asynchronous dependencies can block the event loop or cause runtime errors

8. Global state issues - Not properly managing singleton dependencies leads to shared state bugs in concurrent requests

9. Exception handling - Not handling exceptions in dependencies properly can leave resources in inconsistent states or leak connections

10. Type hint mistakes - Missing or incorrect type hints break dependency injection and automatic validation

11. Ignoring dependency order - Dependencies are executed in the order they appear in function signature; incorrect order can cause issues

12. Not cleaning test overrides - Forgetting to reset app.dependency_overrides after tests causes subsequent tests to fail

13. Overusing global dependencies - Applying too many dependencies globally can hurt performance and make debugging difficult

14. Memory leaks with generators - Not properly closing resources in finally block of generator dependencies causes memory leaks

15. Security misconfiguration - Using weak or missing security dependencies exposes endpoints to unauthorized access

Advanced Caching Patterns

Implement sophisticated caching strategies with dependencies.

from fastapi import Depends
import hashlib
import json

# Multi-layer cache with fallback
class CacheLayer:
    def __init__(
        self,
        memory_cache = Depends(get_memory_cache),
        redis_cache = Depends(get_redis_cache)
    ):
        self.memory = memory_cache
        self.redis = redis_cache

    async def get(self, key: str):
        # Try memory first
        value = self.memory.get(key)
        if value:
            return value

        # Try Redis
        value = await self.redis.get(key)
        if value:
            # Populate memory cache
            self.memory.set(key, value)
            return value

        return None

    async def set(self, key: str, value, ttl: int = 3600):
        self.memory.set(key, value, ttl=min(ttl, 300))
        await self.redis.set(key, value, ttl=ttl)


 ():
    key_data = json.dumps({: args, : kwargs}, sort_keys=)
     hashlib.md5(key_data.encode()).hexdigest()


 ():
      ():
        cache_key = create_cache_key(**params)
        cached_value =  cache.get(cache_key)

         cached_value:
             cached_value

        
        value =  compute_expensive_value(params)
         cache.(cache_key, value, ttl=ttl)
         value

     dependency


  ():
     result


 :
     ():
        .cache = cache
        .invalidation_queue = []

     ():
        .invalidation_queue.append(pattern)

      ():
         pattern  .invalidation_queue:
             .cache.redis.delete_pattern(pattern)
        .invalidation_queue.clear()


  ():
    new_user =  db.create_user(user)
    invalidator.invalidate()
     invalidator.flush()
     new_user

Middleware-Style Dependencies

Use dependencies for cross-cutting concerns.

from fastapi import Depends, Request
from time import time

# Request timing dependency
async def measure_request_time(request: Request):
    start_time = time()
    yield
    duration = time() - start_time
    print(f'{request.method} {request.url.path} took {duration:.3f}s')

# Request ID tracking
async def track_request_id(request: Request):
    request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()))
    request.state.request_id = request_id
    yield request_id

def get_request_id(request: Request):
    return request.state.request_id

# Rate limiting per user
class RateLimiter:
    def __init__(self):
        self.requests = {}

    async def check_rate_limit(
        self,
        user = Depends(get_current_user),
        cache = Depends(get_cache)
    ):
        key = f'rate_limit:{user.id}'
        count = await cache.incr(key)

        if count == 1:
             cache.expire(key, )

         count > :
             HTTPException(
                status_code=,
                detail=
            )
         

rate_limiter = RateLimiter()


  ():
     {: }


  ():
    content_type = request.headers.get()
      content_type     content_type:
         HTTPException(
            status_code=,
            detail=
        )
     


  ():
     data

Resources

Official FastAPI documentation and guides:

• FastAPI Dependencies Core dependency injection concepts and basic usage
• Advanced Dependencies Advanced patterns including dependency overrides and custom providers
• Sub-Dependencies Building complex dependency chains and hierarchies
• Dependencies in Path Operations Using dependencies at the decorator level
• Global Dependencies Applying dependencies to entire applications or routers
• Testing Dependencies Overriding dependencies for testing purposes
• Security and OAuth2 Security utilities and OAuth2 implementation with dependencies
• SQL Databases Database session management with dependency injection
• Async SQL Async database patterns with SQLAlchemy

Additional resources:

• FastAPI GitHub Source code and examples
• FastAPI Discussions Community questions and patterns
• Dependency Injection Design Pattern General DI concepts and theory