Production Tips

Project structure, exception handling, logging, and OpenAPI customization patterns used in real-world FastAPI projects.

---

Project Structure

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app entry point
│   ├── config.py            # Environment settings (Pydantic Settings)
│   ├── dependencies.py      # Common dependencies (DB session, current user)
│   ├── exceptions.py        # Custom exceptions
│   ├── models/              # SQLAlchemy ORM models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic request/response schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── crud/                # DB access layer
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── routers/             # APIRouter collection
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   └── utils/               # Utility functions
│       └── security.py
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── alembic/                 # DB migrations
├── pyproject.toml
└── .env

---

Environment Settings (Pydantic Settings)

# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    app_name: str = "My API"
    debug: bool = False
    version: str = "1.0.0"

    database_url: str = "sqlite+aiosqlite:///./app.db"

    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    redis_url: str = "redis://localhost:6379"

    model_config = {
        "env_file": ".env",
        "env_file_encoding": "utf-8",
    }


@lru_cache
def get_settings() -> Settings:
    return Settings()


settings = get_settings()
print(settings.database_url)

# .env file
SECRET_KEY=your-super-secret-key-change-me
DATABASE_URL=postgresql+asyncpg://user:pass@localhost/mydb
DEBUG=false

---

Global Exception Handling

# exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import logging

logger = logging.getLogger(__name__)


class AppException(Exception):
    def __init__(self, status_code: int, detail: str, code: str = "ERROR"):
        self.status_code = status_code
        self.detail = detail
        self.code = code


class NotFoundError(AppException):
    def __init__(self, resource: str, resource_id: int | str):
        super().__init__(
            status_code=404,
            detail=f"{resource} {resource_id} not found",
            code="NOT_FOUND",
        )


class ConflictError(AppException):
    def __init__(self, detail: str):
        super().__init__(status_code=409, detail=detail, code="CONFLICT")


def setup_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppException)
    async def app_exception_handler(request: Request, exc: AppException):
        logger.warning(f"AppException: {exc.code} - {exc.detail}")
        return JSONResponse(
            status_code=exc.status_code,
            content={"code": exc.code, "detail": exc.detail},
        )

    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        errors = [
            {
                "field": ".".join(str(loc) for loc in error["loc"][1:]),
                "message": error["msg"],
            }
            for error in exc.errors()
        ]
        return JSONResponse(
            status_code=422,
            content={"code": "VALIDATION_ERROR", "errors": errors},
        )

    @app.exception_handler(Exception)
    async def general_exception_handler(request: Request, exc: Exception):
        logger.error(f"Unexpected error: {exc}", exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"code": "INTERNAL_ERROR", "detail": "An internal server error occurred"},
        )

---

Structured Logging

# logging_config.py
import logging
import json
from datetime import datetime


class JSONFormatter(logging.Formatter):
    """JSON log format (for log aggregation systems)"""

    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)


def setup_logging(debug: bool = False):
    level = logging.DEBUG if debug else logging.INFO

    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())

    logging.basicConfig(level=level, handlers=[handler])

    # Reduce noise from external libraries
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

---

OpenAPI Customization

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def custom_openapi(app: FastAPI):
    if app.openapi_schema:
        return app.openapi_schema

    schema = get_openapi(
        title="My Production API",
        version="2.0.0",
        description="""
## Authentication
All protected endpoints require `Authorization: Bearer <token>` header.

## Error Codes
- `NOT_FOUND`: Resource not found
- `CONFLICT`: Duplicate data
- `VALIDATION_ERROR`: Input validation failed
        """,
        routes=app.routes,
        tags=[
            {"name": "users", "description": "User management"},
            {"name": "items", "description": "Item management"},
            {"name": "auth", "description": "Authentication"},
        ],
    )

    # Add JWT security scheme
    schema["components"]["securitySchemes"] = {
        "BearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT",
        }
    }

    app.openapi_schema = schema
    return schema


app = FastAPI(docs_url="/docs", redoc_url="/redoc")
app.openapi = lambda: custom_openapi(app)

---

Final main.py Assembly

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import get_settings
from exceptions import setup_exception_handlers
from logging_config import setup_logging
from database import create_tables
from routers import users, items, auth

settings = get_settings()
setup_logging(debug=settings.debug)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_tables()
    yield


app = FastAPI(
    title=settings.app_name,
    version=settings.version,
    lifespan=lifespan,
    docs_url="/docs" if settings.debug else None,  # hide docs in production
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)

setup_exception_handlers(app)

app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])


@app.get("/health")
def health_check():
    return {"status": "ok", "version": settings.version}

---

Summary

Pattern	Description
Layered structure	routers → crud → models
Pydantic Settings	.env-based environment config
Global exception handlers	Consistent error response format
JSON logging	Integration with log aggregation
OpenAPI customization	Better docs, security schema
lifespan	App startup/shutdown event management

In production, separation of concerns (routers, CRUD, schemas, models) and per-environment configuration are the keys to maintainability.