FastAPI Production Best Practices: Build Bulletproof Python APIs (2026)

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.