FastAPI: Production Architecture & Patterns

FastAPI is deceptively simple to start but demands real architecture to scale. This chapter covers patterns used in production: layered app structure, middleware composition, dependency injection beyond docs examples, and background task patterns.

Production App Structure

The single-file main.py works for demos. Production apps need separation of concerns:

project/
├── app/
│   ├── main.py              # App factory, lifespan events
│   ├── config.py             # Settings from env vars
│   ├── dependencies.py       # Shared DI functions
│   ├── middleware/
│   │   ├── logging.py
│   │   ├── rate_limit.py
│   │   └── request_id.py
│   ├── routers/
│   │   ├── users.py
│   │   ├── orders.py
│   │   └── health.py
│   ├── models/
│   │   ├── user.py           # SQLModel / SQLAlchemy
│   │   └── order.py
│   ├── schemas/
│   │   ├── user.py           # Pydantic request/response
│   │   └── order.py
│   └── services/
│       ├── user_service.py   # Business logic
│       └── order_service.py
├── alembic/                  # Database migrations
├── tests/
└── docker-compose.yml

Application Factory Pattern

Don't create app = FastAPI() at module level. Use a factory:

# app/main.py
def create_app() -> FastAPI:
    app = FastAPI(title="My API", version="2.0.0")
    
    # Register middleware (order matters!)
    app.add_middleware(RequestIDMiddleware)
    app.add_middleware(LoggingMiddleware)
    app.add_middleware(RateLimitMiddleware)
    
    # Register routers
    app.include_router(users.router, prefix="/users", tags=["users"])
    app.include_router(orders.router, prefix="/orders", tags=["orders"])
    app.include_router(health.router, prefix="/health", tags=["health"])
    
    return app

Structured Error Handling

FastAPI has exception handlers. Use them consistently instead of try/except in every endpoint:

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    """Base exception for application errors."""
    def __init__(self, message: str, status_code: int = 400, code: str = "APP_ERROR"):
        self.message = message
        self.status_code = status_code
        self.code = code

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

class ValidationError(AppException):
    def __init__(self, details: str):
        super().__init__(message=details, status_code=422, code="VALIDATION_ERROR")

# Global exception handler
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "request_id": request.state.request_id
            }
        }
    )

Background Tasks

Don't make users wait for non-critical work like sending emails or generating reports:

Lifespan Events — Startup & Shutdown

Connect to your database and message broker on startup, clean up on shutdown:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await database.connect()
    await redis.initialize()
    logger.info("Application started")
    
    yield  # Application runs here
    
    # Shutdown
    await database.disconnect()
    await redis.close()
    logger.info("Application shutting down")

app = FastAPI(lifespan=lifespan)

Quiz

Key Takeaways

✅ Factory pattern for testable, configurable app instances.
✅ Layered architecture: routers → services → models keeps code organized.
✅ Middleware runs in order — design your pipeline intentionally.
✅ Depends() builds a dependency graph, not just a function call.
✅ Background tasks keep API responses fast for non-critical work.
✅ Lifespan events for proper startup/shutdown resource management.

Keep coding! 🚀