FastAPI Advanced - Background Tasks, Dependencies

Introduction

FastAPI provides advanced features including background tasks for async processing and dependency injection for modular code. These features enable scalable and maintainable applications.

Background Tasks

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import time

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str

def process_email(email: str, subject: str):
    """Background task function."""
    time.sleep(5)
    print(f"Email sent to {email}: {subject}")

def process_data(data: dict):
    """Process data in background."""
    time.sleep(10)
    result = data['value'] * 2
    print(f"Processed result: {result}")

@app.post("/items/")
async def create_item(item: Item, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_email, "user@example.com", f"Item created: {item.name}")
    return {"message": "Item created, email will be sent"}

@app.post("/process/")
async def process_data_endpoint(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data, data)
    return {"message": "Processing started"}

Dependency Injection

from fastapi import Depends, FastAPI, HTTPException
from typing import Optional
from functools import lru_cache

app = FastAPI()

# Simple dependency
async def get_db():
    db = DatabaseConnection()
    try:
        yield db
    finally:
        db.close()

# Dependency with parameters
def get_settings():
    return Settings()

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

# Class-based dependency
class PaginationParams:
    def __init__(self, page: int = 1, size: int = 10):
        if page < 1:
            raise HTTPException(status_code=400, detail="Page must be >= 1")
        if size > 100:
            raise HTTPException(status_code=400, detail="Size must be <= 100")
        self.page = page
        self.size = size

# Dependency with sub-dependencies
async def get_current_user(token: str = Depends(get_token)):
    return decode_token(token)

@app.get("/items/")
async def list_items(
    db = Depends(get_db),
    pagination: PaginationParams = Depends(),
    current_user: User = Depends(get_current_user)
):
    return db.get_items(pagination.page, pagination.size)

Advanced Dependencies

# Conditional dependencies
def get_optional_db():
    return DatabaseConnection()

@app.get("/items/")
async def list_items(db: DatabaseConnection = Depends(get_optional_db, None)):
    if db is None:
        return []
    return db.get_all()

# Multiple dependencies
async def verify_token(token: str = Depends(get_token)):
    return token

async def verify_api_key(api_key: str = Depends(get_api_key)):
    return api_key

@app.post("/items/")
async def create_item(
    token: str = Depends(verify_token),
    api_key: str = Depends(verify_api_key),
    item: Item
):
    return {"item": item}

# Dependency override for testing
app.dependency_overrides[get_db] = get_test_db

Practice Problems

1. Create a background task that sends a webhook notification after an operation
2. Implement a dependency that provides rate limiting functionality
3. Build a dependency injection chain with authentication and authorization
4. Add a dependency that caches results for expensive operations
5. Create a dependency that handles database transactions