Dependency Injection: The Right Way
FastAPI’s dependency injection is its most powerful feature and most commonly misused. Dependencies should be composable, testable, and cover cross-cutting concerns like auth, database sessions, and rate limiting.
# app/core/dependencies.py
from typing import Annotated, AsyncGenerator
from uuid import UUID
from fastapi import Depends, Header, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.db.session import AsyncSessionLocal
from app.models.user import User
from app.services.user_service import UserService
# --- Database session dependency (async, scoped per request) ---
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
DBSession = Annotated[AsyncSession, Depends(get_db)]
# --- JWT auth dependency ---
security = HTTPBearer(auto_error=False)
async def get_current_user(
db: DBSession,
credentials: HTTPAuthorizationCredentials | None = Depends(security),
) -> User:
if credentials is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="No authentication credentials provided",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
credentials.credentials,
settings.SECRET_KEY,
algorithms=[settings.ALGORITHM],
)
user_id: str | None = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid token payload")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid or expired token")
user_service = UserService(db)
user = await user_service.get_by_id(UUID(user_id))
if user is None or not user.is_active:
raise HTTPException(status_code=401, detail="User not found or inactive")
return user
CurrentUser = Annotated[User, Depends(get_current_user)]
# --- Role-based access control (composable with CurrentUser) ---
def require_role(*roles: str):
async def check_role(current_user: CurrentUser) -> User:
if current_user.role not in roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Role '{current_user.role}' is not authorized for this action",
)
return current_user
return check_role
AdminUser = Annotated[User, Depends(require_role("admin"))]
# --- Pagination dependency ---
class PaginationParams:
def __init__(
self,
page: Annotated[int, Query(ge=1, default=1)],
page_size: Annotated[int, Query(ge=1, le=100, default=20)],
):
self.offset = (page - 1) * page_size
self.limit = page_size
self.page = page
self.page_size = page_size
Pagination = Annotated[PaginationParams, Depends(PaginationParams)]
Structured Error Handling
FastAPI’s default error responses are inconsistent — validation errors use one shape, HTTP exceptions use another. Production APIs need a uniform error envelope that API clients can reliably parse.
# app/core/error_handlers.py
import logging
import traceback
from typing import Any
from uuid import uuid4
from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from sqlalchemy.exc import IntegrityError
logger = logging.getLogger(__name__)
class APIError(Exception):
"""Base class for all API errors."""
def __init__(
self,
message: str,
status_code: int = 400,
error_code: str = "API_ERROR",
details: dict[str, Any] | None = None,
):
self.message = message
self.status_code = status_code
self.error_code = error_code
self.details = details or {}
super().__init__(message)
class NotFoundError(APIError):
def __init__(self, resource: str, resource_id: Any):
super().__init__(
message=f"{resource} with id '{resource_id}' not found",
status_code=404,
error_code="NOT_FOUND",
details={"resource": resource, "id": str(resource_id)},
)
class ConflictError(APIError):
def __init__(self, message: str):
super().__init__(message=message, status_code=409, error_code="CONFLICT")
def error_response(
request: Request,
status_code: int,
error_code: str,
message: str,
details: dict | None = None,
) -> JSONResponse:
return JSONResponse(
status_code=status_code,
content={
"error": {
"code": error_code,
"message": message,
"details": details or {},
"request_id": getattr(request.state, "request_id", str(uuid4())),
"path": str(request.url.path),
}
},
)
def register_error_handlers(app: FastAPI) -> None:
@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError) -> JSONResponse:
return error_response(request, exc.status_code, exc.error_code, exc.message, exc.details)
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
field_errors = {}
for error in exc.errors():
field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
field_errors[field] = error["msg"]
return error_response(
request, 422, "VALIDATION_ERROR", "Request validation failed", {"fields": field_errors}
)
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError) -> JSONResponse:
logger.warning("DB integrity error: %s", exc, extra={"request_id": getattr(request.state, "request_id", "")})
return error_response(request, 409, "CONFLICT", "Resource already exists")
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
request_id = getattr(request.state, "request_id", str(uuid4()))
logger.error(
"Unhandled exception",
exc_info=True,
extra={"request_id": request_id, "path": str(request.url.path)},
)
return error_response(
request, 500, "INTERNAL_ERROR",
"An unexpected error occurred",
{"request_id": request_id},
)
Middleware: Request ID + Timing
# app/middleware/request_id.py
import time
from uuid import uuid4
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
request_id = request.headers.get("X-Request-ID", str(uuid4()))
request.state.request_id = request_id
start_time = time.perf_counter()
response = await call_next(request)
duration_ms = (time.perf_counter() - start_time) * 1000
response.headers["X-Request-ID"] = request_id
response.headers["X-Response-Time"] = f"{duration_ms:.2f}ms"
return response
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from app.api.v1.router import api_router
from app.core.error_handlers import register_error_handlers
from app.core.logging import configure_logging
from app.middleware.request_id import RequestIDMiddleware
@asynccontextmanager
async def lifespan(app: FastAPI):
configure_logging()
# Startup: initialize connection pools, warm caches
yield
# Shutdown: close connections gracefully
def create_app() -> FastAPI:
app = FastAPI(
title="My Production API",
version="1.0.0",
docs_url="/docs" if settings.ENVIRONMENT != "production" else None,
redoc_url=None,
lifespan=lifespan,
)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
register_error_handlers(app)
app.include_router(api_router, prefix="/api/v1")
return app
app = create_app()
Background Tasks With Error Handling
FastAPI’s built-in BackgroundTasks runs tasks in the same process after the response is sent. For anything CPU-heavy or long-running, use Celery or ARQ instead. For lightweight post-response work (sending emails, logging, cache invalidation), BackgroundTasks is fine — but you must handle exceptions or they silently disappear.
# app/services/background.py
import logging
from typing import Callable, Any
from functools import wraps
logger = logging.getLogger(__name__)
def background_task(func: Callable) -> Callable:
"""Decorator that wraps background tasks with exception logging."""
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> None:
try:
await func(*args, **kwargs)
except Exception:
logger.exception(
"Background task failed: %s",
func.__name__,
extra={"args": str(args)[:200], "kwargs": str(kwargs)[:200]},
)
return wrapper
# Usage in an endpoint:
# app/api/v1/endpoints/users.py
from fastapi import APIRouter, BackgroundTasks, status
from app.core.dependencies import DBSession
from app.schemas.user import UserCreate, UserRead
from app.services.background import background_task
from app.services.email_service import send_welcome_email
from app.services.user_service import UserService
router = APIRouter(prefix="/users", tags=["users"])
@background_task
async def send_welcome_email_task(user_id: str, email: str, full_name: str) -> None:
await send_welcome_email(user_id=user_id, email=email, full_name=full_name)
@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(
payload: UserCreate,
background_tasks: BackgroundTasks,
db: DBSession,
) -> UserRead:
service = UserService(db)
user = await service.create(payload)
background_tasks.add_task(
send_welcome_email_task,
user_id=str(user.id),
email=user.email,
full_name=user.full_name,
)
return user
Testing: Async pytest With Real Dependencies
# tests/conftest.py
import asyncio
from typing import AsyncGenerator
from uuid import uuid4
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.core.dependencies import get_db
from app.db.base import Base
from app.main import app
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
@pytest_asyncio.fixture(scope="session")
async def setup_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def db_session(setup_db) -> AsyncGenerator[AsyncSession, None]:
async with TestSessionLocal() as session:
yield session
await session.rollback()
@pytest_asyncio.fixture
async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
yield ac
app.dependency_overrides.clear()
# tests/test_users.py
import pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_create_user_success(client: AsyncClient):
response = await client.post("/api/v1/users/", json={
"email": "[email protected]",
"full_name": "Test User",
"password": "Secure@123",
"password_confirm": "Secure@123",
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "[email protected]"
assert "id" in data
assert "password" not in data # Never leak passwords
@pytest.mark.asyncio
async def test_create_user_weak_password(client: AsyncClient):
response = await client.post("/api/v1/users/", json={
"email": "[email protected]",
"full_name": "Test User",
"password": "weakpassword",
"password_confirm": "weakpassword",
})
assert response.status_code == 422
error = response.json()["error"]
assert error["code"] == "VALIDATION_ERROR"
@pytest.mark.asyncio
async def test_create_user_duplicate_email(client: AsyncClient):
payload = {
"email": "[email protected]",
"full_name": "Dup User",
"password": "Strong@123",
"password_confirm": "Strong@123",
}
await client.post("/api/v1/users/", json=payload)
response = await client.post("/api/v1/users/", json=payload)
assert response.status_code == 409
People Also Ask
Is FastAPI ready for production in 2026?
Yes. FastAPI is production-ready and widely deployed at companies like Microsoft, Uber, and Netflix. The key additions for production are structured error handling, async SQLAlchemy, proper dependency injection, and a test suite with async pytest. The defaults out of the box are not production-grade.
Should I use Pydantic v1 or v2 with FastAPI?
Pydantic v2 is the standard in 2026. FastAPI 0.100+ ships with Pydantic v2 by default. The validator API changed significantly — use @field_validator and @model_validator instead of the v1 @validator. The performance improvement (5-50x faster validation) is meaningful at API scale.
How do I handle database sessions safely in async FastAPI?
Use a dependency that yields from an async context manager, with explicit commit on success and rollback on exception. Never share a session across requests. The pattern shown above — async with AsyncSessionLocal() as session with try/except/rollback — is the correct production pattern with SQLAlchemy 2.0+ async.
For production-ready developer tools, templates, and starter kits including FastAPI boilerplates, see WOWHOW developer tools. To explore all Python and backend resources, browse the full catalog.
Comments · 0
Beta: comments are stored locally on your device and not visible to other readers.
No comments yet. Be the first to share your thoughts.