mkdir catalog-service && cd catalog-service
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pydantic-settings sqlalchemy aiosqlite

from pydantic import BaseModel, Field
from typing import Optional

class ProductSchema(BaseModel):
    sku: str = Field(..., min_length=3, max_length=20)
    title: str
    unit_price: float = Field(..., gt=0.0)
    inventory_count: int = Field(default=0, ge=0)
    category: Optional[str] = None

class ProductResponse(ProductSchema):
    id: int

from fastapi import APIRouter, Depends, HTTPException, Query
from typing import List

router = APIRouter(prefix="/catalog", tags=["Inventory"])

# Simulated persistence layer for demonstration
_in_memory_store: dict[int, ProductResponse] = {}
_next_id: int = 1

def get_store():
    return _in_memory_store

@router.post("/products", response_model=ProductResponse, status_code=201)
async def register_product(payload: ProductSchema, store: dict = Depends(get_store)):
    global _next_id
    new_record = ProductResponse(id=_next_id, **payload.model_dump())
    store[_next_id] = new_record
    _next_id += 1
    return new_record

@router.get("/products", response_model=List[ProductResponse])
async def fetch_products(
    store: dict = Depends(get_store),
    category: Optional[str] = Query(None),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100)

): filtered = store.values() if category: filtered = [p for p in filtered if p.category == category]

start = (page - 1) * page_size
end = start + page_size
return list(filtered)[start:end]

@router.get("/products/{product_id}", response_model=ProductResponse) async def retrieve_product(product_id: int, store: dict = Depends(get_store)): if product_id not in store: raise HTTPException(status_code=404, detail=f"Product {product_id} not indexed") return store[product_id]

@router.delete("/products/{product_id}", status_code=204) async def decommission_product(product_id: int, store: dict = Depends(get_store)): if product_id not in store: raise HTTPException(status_code=404, detail="Record missing") del store[product_id]

### Step 4: Application Assembly & Middleware

The main application file acts as a composition root. CORS, logging, and lifespan events are registered here to maintain separation of concerns.

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .routers import catalog

app = FastAPI(title="Inventory Control API", version="2.1.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://admin.internal.corp"],
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

app.include_router(catalog.router)

Architecture Rationale

• Schema-First Routing: Pydantic models serve as single sources of truth for validation, serialization, and OpenAPI generation. This eliminates manual request.json() parsing and reduces runtime type errors.
• Dependency Injection via Depends: Resources like database sessions, configuration objects, and external clients are injected per-request. This prevents global state leaks, simplifies testing, and enables automatic cleanup.
• Modular APIRouter: Domain segmentation keeps route handlers focused. Each router can be tested in isolation and mounted conditionally based on environment or feature flags.
• Explicit Status Codes & Response Models: Declaring response_model and status_code enforces contract stability. Clients receive predictable payloads, and automated documentation stays synchronized with implementation.

Pitfall Guide

1. Blocking the Event Loop with CPU-Bound Tasks

Explanation: FastAPI routes are async by default. Running heavy computations, synchronous database drivers, or blocking I/O inside an async def handler stalls the entire event loop, degrading throughput for all concurrent requests. Fix: Offload CPU-intensive work to run_in_threadpool or use asyncio.to_thread. For database operations, always use async drivers (asyncpg, aiosqlite, databases).

2. Pydantic v2 Migration Traps

Explanation: Upgrading from Pydantic v1 breaks code using .dict(), .json(), or the inner class Config. These patterns raise AttributeError or silently misbehave. Fix: Replace .dict() with .model_dump(), .json() with .model_dump_json(), and move configuration to model_config = ConfigDict(...). Run pydantic-migrate or audit imports before deployment.

3. In-Memory State in Multi-Worker Deployments

Explanation: Storing data in module-level dictionaries works locally but fails when Uvicorn spawns multiple workers. Each process maintains an isolated copy, causing data inconsistency and lost writes. Fix: Externalize state to Redis, PostgreSQL, or a message queue. Use connection pooling and idempotent write patterns. Reserve in-memory stores strictly for prototyping or caching layers with TTL.

4. Overly Permissive CORS Configuration

Explanation: Setting allow_origins=["*"] with allow_credentials=True creates a security vulnerability. Browsers will reject the combination, but misconfigured headers can still expose endpoints to cross-site request forgery. Fix: Explicitly whitelist trusted origins. Separate frontend and backend CORS policies. Use environment variables to inject allowed domains per deployment stage.

5. Ignoring Depends for Resource Lifecycle Management

Explanation: Opening database connections or HTTP clients inside route handlers without cleanup leads to connection leaks and exhausted pools under load. Fix: Wrap resource acquisition in a generator function yielded via Depends. FastAPI automatically handles teardown after the response is sent.

async def get_db_session():
    session = await engine.connect()
    try:
        yield session
    finally:
        await session.close()

6. Manual Query Parameter Parsing

Explanation: Extracting ?skip=10&limit=5 via request.query_params bypasses FastAPI's validation pipeline. Invalid types or missing bounds cause runtime crashes instead of structured 422 responses. Fix: Declare query parameters as function arguments with type hints and Query() constraints. The framework handles parsing, validation, and documentation automatically.

7. Data Leakage Through Missing Response Models

Explanation: Returning raw database objects or dictionaries exposes internal fields (passwords, internal IDs, audit flags) to clients. Fix: Always specify response_model. Use Pydantic's Field(exclude=True) or create dedicated response schemas that strip sensitive attributes before serialization.

Production Bundle

Action Checklist

• Pin dependencies: Lock fastapi, uvicorn, and pydantic versions in requirements.txt or pyproject.toml
• Enable async drivers: Replace synchronous DB libraries with asyncpg, aiosqlite, or SQLAlchemy 2.0 async
• Configure lifespan events: Use @app.on_event("startup") or asynccontextmanager for connection pooling and cache warming
• Enforce response contracts: Attach response_model to every endpoint to prevent schema drift
• Implement structured logging: Replace print() with logging or structlog for request tracing and error aggregation
• Add health checks: Expose /health and /ready endpoints for container orchestration and load balancer probes
• Validate CORS boundaries: Restrict origins to known frontend domains; never use wildcards in production

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
High-concurrency REST API	FastAPI + Uvicorn	Native ASGI, automatic validation, ~15k req/s baseline	Low infrastructure cost, high dev velocity
Legacy monolith migration	Django REST Framework	Built-in admin, ORM maturity, gradual migration path	Higher memory footprint, slower iteration
Simple webhook processor	Flask + lightweight routing	Minimal overhead, straightforward deployment	Manual docs/validation adds maintenance cost
Full-stack SaaS with auth	Django + DRF	Integrated auth, admin panel, battle-tested patterns	Higher initial setup, slower async adoption
Real-time WebSocket service	FastAPI + WebSockets or aiohttp	Shared ASGI runtime, easy routing integration	Requires connection state management

Configuration Template

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class AppSettings(BaseSettings):
    app_name: str = "Catalog Service"
    debug: bool = False
    database_url: str
    allowed_origins: list[str] = ["https://app.internal.corp"]
    
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = AppSettings()

# main.py
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import settings
from routers import catalog

app = FastAPI(title=settings.app_name, version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
)

app.include_router(catalog.router)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=settings.debug)

Quick Start Guide

1. Initialize environment: Create a virtual environment and install fastapi, uvicorn, and pydantic-settings.
2. Define schemas: Write Pydantic models for request payloads and response contracts. Use Field() constraints for validation boundaries.
3. Mount router: Create an APIRouter, attach endpoints with type-hinted parameters, and include it in the main FastAPI instance.
4. Launch server: Run uvicorn main:app --reload and navigate to http://localhost:8000/docs to verify auto-generated OpenAPI documentation.
5. Validate contracts: Test endpoints with curl or the interactive Swagger UI. Confirm that invalid payloads return 422 Unprocessable Entity with structured error details.