Dependency Injection & Async

Depends() — Dependency Injection

from fastapi import Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

# Simple dependency
def common_pagination(page: int = 1, limit: int = 10):
    return {"skip": (page - 1) * limit, "limit": limit}

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

# Auth dependency
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = await db.get_user(user_id)
    if user is None:
        raise credentials_exception
    return user

# Use auth dependency
@app.get("/profile")
async def get_profile(current_user: User = Depends(get_current_user)):
    return current_user

# Database session dependency
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

Async Route Handlers

import asyncio
import httpx
from fastapi import BackgroundTasks

# Async route — use for I/O bound work (DB, HTTP, files)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await db.fetch_user(user_id)   # awaitable DB call
    return user

# Sync route — use for CPU bound work; runs in threadpool
@app.get("/process")
def process_data():
    result = heavy_cpu_task()            # runs in threadpool automatically
    return result

# Background tasks
@app.post("/email")
async def send_email(
    email: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(send_welcome_email, email)  # runs after response
    return {"message": "Email queued"}

# Parallel async calls
@app.get("/dashboard")
async def dashboard(user_id: int):
    users_task = asyncio.create_task(db.get_user(user_id))
    orders_task = asyncio.create_task(db.get_orders(user_id))
    stats_task = asyncio.create_task(db.get_stats(user_id))
    user, orders, stats = await asyncio.gather(users_task, orders_task, stats_task)
    return {"user": user, "orders": orders, "stats": stats}

# App lifespan — startup/shutdown
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await db.connect()
    await redis.connect()
    yield
    # Shutdown
    await db.disconnect()
    await redis.disconnect()

app = FastAPI(lifespan=lifespan)

SQLAlchemy Async + Alembic

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=False, pool_size=10, max_overflow=20)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# models.py
from sqlalchemy import String, Boolean, Integer, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    name: Mapped[str] = mapped_column(String)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    orders: Mapped[list["Order"]] = relationship(back_populates="user")

# CRUD with async SQLAlchemy
from sqlalchemy import select, update, delete

async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100):
    result = await db.execute(select(User).offset(skip).limit(limit))
    return result.scalars().all()

async def create_user(db: AsyncSession, email: str, name: str) -> User:
    user = User(email=email, name=name)
    db.add(user)
    await db.flush()   # get ID without committing
    return user

Middleware & Exception Handlers

from fastapi import Request
from fastapi.responses import JSONResponse
import time

# Middleware
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response

# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

# Custom exception
class AppError(Exception):
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message},
    )