fastapi-dependency-injection

What this skill does

# 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.

```python
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.

```python
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 = "MyApp"
        self.admin_email = "[email protected]"

settings = Settings()  # Created once at startup

def get_settings():
    return settings

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

## Sub-Dependencies

Build complex dependency chains.

```python
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 delete_user(
    user_id: int,
    admin = Depends(get_admin_user),
    db = Depends(get_db)
):
    await db.execute('DELETE FROM users WHERE id = ?', user_id)
    return {'status': 'deleted'}
```

## Class-Based Dependencies

Use classes for stateful dependencies.

```python
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 not self.db:
            self.db = Database('postgresql://localhost/db')
            await self.db.connect()
        return self.db

db_dependency = DatabaseDependency()

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

# Class with initialization parameters
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = limit

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

# Advanced class-based dependency with configuration
class ServiceClient:
    def __init__(
        self,
        api_key: str,
        timeout: int = 30,
        max_retries: int = 3
    ):
        self.api_key = api_key
        self.timeout = timeout
        self.max_retries = max_retries
        self.client = None

    async def __call__(self):
        if not self.client:
            self.client = await create_http_client(
                api_key=self.api_key,
                timeout=self.timeout
            )
        return self.client

# Factory function for configurable class-based dependency
def create_service_dependency(api_key: str):
    return ServiceClient(api_key=api_key, timeout=60)

service = create_service_dependency('my-api-key')

@app.get('/external-data')
async def get_external_data(client = Depends(service)):
    return await client.fetch('/data')
```

## Generator Dependencies for Cleanup

Use generator functions to ensure proper resource cleanup.

```python
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:
        yield file
    finally:
        await file.close()

# Multiple resources with cleanup
async def get_resources():
    db = await Database.connect('postgresql://localhost/db')
    cache = await Redis.connect('redis://localhost')
    logger = Logger('app')

    try:
        yield {'db': db, 'cache': cache, 'logger': logger}
    finally:
        await cache.close()
        await db.close()
        logger.shutdown()

@app.post('/process')
async def process_data(
    data: dict,
    resources = Depends(get_resources)
):
    db = resources['db']
    cache = resources['cache']
    logger = resources['logger']

    logger.info('Processing data')
    result = await db.save(data)
    await cache.invalidate('data_cache')
    return result

# Context manager as dependency
@asynccontextmanager
async def transaction_context(db = Depends(get_db)):
    async with db.begin() as transaction:
        try:
            yield transaction
            await transaction.commit()
        except Exception:
            await transaction.rollback()
            raise

async def get_transaction(db = Depends(get_db)):
    async with transaction_context(db) as txn:
        yield txn

@app.post('/transfer')
async def transfer_funds(
    from_account: int,
    to_account: int,
    amount: float,
    txn = Depends(get_transaction)
):
    await txn.execute(
        'UPDATE accounts SET balance = balance - ? WHERE id = ?',
        amount, from_account
    )
    await txn.execute(
        'UPDATE accounts