sku: str = Field(..., min_length=3, max_length=20, pattern=r"^[A-Z0-9]+$")
quantity: int = Field(..., gt=0)
location_code: Optional[str] = Field(None, max_length=10)
class InventoryOut(BaseModel):
id: int
sku: str
quantity: int
location_code: Optional[str]
updated_at: datetime
**Why this structure:** Separating input and output models prevents accidental data leakage and allows independent evolution of internal storage formats versus external contracts. `ConfigDict` replaces the deprecated `class Config` pattern, and `Field` constraints move validation logic out of route handlers.
### 2. Dependency Injection for Resource Management
FastAPI's `Depends` system manages lifecycle hooks cleanly. Use it for database sessions, authentication contexts, and configuration injection.
```python
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/warehouse"
engine = create_async_engine(DATABASE_URL)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_transaction_session() -> AsyncSession:
async with async_session() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def verify_admin_token(token: str = Depends(get_auth_header)) -> dict:
if token != "VALID_ADMIN_TOKEN":
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
return {"role": "admin"}
Why this approach: Generator-based dependencies (yield) guarantee cleanup even when exceptions occur. Transaction boundaries are explicitly managed, preventing connection leaks under load. Authentication is decoupled from business logic, enabling straightforward unit testing.
3. Async Route Handlers & Background Processing
I/O-bound operations should never block the event loop. Use async def for database calls and external API requests. Offload non-critical work to BackgroundTasks.
from fastapi import APIRouter, Depends, BackgroundTasks
from sqlalchemy import select
router = APIRouter(prefix="/inventory", tags=["stock-management"])
@router.post("/restock", response_model=InventoryOut, status_code=201)
async def add_stock(
payload: InventoryIn,
db: AsyncSession = Depends(get_transaction_session),
admin: dict = Depends(verify_admin_token),
background: BackgroundTasks = None
):
query = select(InventoryRecord).where(InventoryRecord.sku == payload.sku)
result = await db.execute(query)
record = result.scalar_one_or_none()
if record:
record.quantity += payload.quantity
record.location_code = payload.location_code or record.location_code
else:
record = InventoryRecord(**payload.model_dump())
db.add(record)
# Flush to generate ID before background task
await db.flush()
if background:
background.add_task(log_inventory_change, record.sku, payload.quantity)
await db.refresh(record)
return record
Why async here: AsyncSession leverages asyncpg for non-blocking PostgreSQL communication. The route handler yields control during database I/O, allowing the event loop to serve other requests. BackgroundTasks executes after the HTTP response is sent, preventing latency spikes from email dispatches or audit logging.
4. Router Composition & Application Assembly
Avoid monolithic main.py files. Group endpoints by domain and mount them at the application level.
# app/routers/inventory.py (above)
# app/routers/reports.py
# app/main.py
from fastapi import FastAPI
from app.routers import inventory, reports
app = FastAPI(
title="Warehouse Management API",
version="2.1.0",
docs_url="/api/docs",
redoc_url="/api/redoc"
)
app.include_router(inventory.router)
app.include_router(reports.router)
Why modular routing: Prefixes and tags auto-organize the generated OpenAPI specification. Domain isolation simplifies permission scoping and enables independent deployment strategies in microservice architectures.
Pitfall Guide
1. Blocking the Event Loop in Async Handlers
Explanation: Using synchronous libraries (e.g., requests, psycopg2, or CPU-heavy computations) inside async def routes blocks the entire event loop, degrading throughput to single-threaded performance.
Fix: Replace synchronous I/O with async equivalents (httpx, asyncpg, aiomysql). For CPU-bound work, offload to Celery, RQ, or run_in_executor.
2. Pydantic v2 Migration Traps
Explanation: Upgrading from v1 to v2 breaks code using .dict(), .json(), or class Config. The validation engine was rewritten in Rust, changing method names and configuration syntax.
Fix: Use .model_dump(), .model_dump_json(), and model_config = ConfigDict(...). Run pydantic-migrate or manually audit schema definitions before deployment.
3. Over-Engineering Dependency Chains
Explanation: Nesting Depends calls too deeply creates opaque execution flows and makes testing difficult. Circular dependencies or shared mutable state across dependencies cause unpredictable behavior.
Fix: Keep dependency graphs shallow (max 2-3 levels). Use explicit parameter passing for simple cases. Mock dependencies at the router level during tests rather than patching global state.
4. Ignoring Background Task Failure Modes
Explanation: BackgroundTasks runs in-process and lacks retry logic, dead-letter queues, or persistence. If the worker crashes mid-execution, tasks are lost silently.
Fix: Use BackgroundTasks only for idempotent, non-critical operations (e.g., cache invalidation, analytics pings). For critical workflows, implement a dedicated task queue (Celery, ARQ, or Temporal).
Explanation: Setting allow_origins=["*"] with allow_credentials=True violates browser security policies and exposes endpoints to cross-site request forgery.
Fix: Explicitly whitelist frontend domains. Use environment variables for origin lists. Validate Origin headers against a allowlist before responding.
6. Testing Without Database Isolation
Explanation: Running integration tests against a shared development database causes race conditions, dirty state, and flaky assertions.
Fix: Spin up a temporary PostgreSQL instance via Testcontainers or SQLite in-memory for tests. Use pytest-asyncio with transactional rollbacks to guarantee clean state per test case.
7. Skipping Request/Response Model Separation
Explanation: Reusing the same Pydantic model for input and output couples validation rules to serialization formats. Internal fields (e.g., password_hash, internal_id) may leak, or required input fields may break response contracts.
Fix: Maintain distinct *In and *Out models. Use inheritance or composition to share common fields while keeping contracts independent.
Production Bundle
Action Checklist
Decision Matrix
| Scenario | Recommended Approach | Why | Cost Impact |
|---|
| High-concurrency I/O workloads | async def + asyncpg/httpx | Non-blocking event loop maximizes throughput per worker | Lower infrastructure cost (fewer instances needed) |
| CPU-heavy data processing | Celery/ARQ + sync FastAPI endpoints | Prevents event loop starvation; enables horizontal scaling | Higher infrastructure cost (worker nodes), but stable API latency |
| Small team / rapid prototyping | Single-file router + SQLite + BackgroundTasks | Minimizes boilerplate; fast iteration cycle | Low initial cost; scales poorly beyond 500 req/s |
| Enterprise microservices | Domain routers + async SQLAlchemy + dedicated task queue | Enforces boundaries; supports independent deployment | Higher initial complexity; reduces long-term maintenance cost |
Configuration Template
# pyproject.toml
[project]
name = "warehouse-api"
version = "2.1.0"
requires-python = ">=3.11"
dependencies = [
"fastapi>=0.109.0",
"uvicorn[standard]>=0.27.0",
"pydantic>=2.6.0",
"sqlalchemy[asyncio]>=2.0.27",
"asyncpg>=0.29.0",
"httpx>=0.27.0",
"python-multipart>=0.0.9",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.23.0",
"testcontainers>=4.5.0",
"ruff>=0.3.0",
]
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
# app/main.py
import os
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import inventory, reports
from app.config import Settings
settings = Settings()
app = FastAPI(
title=settings.APP_NAME,
version=settings.APP_VERSION,
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json" if settings.ENV != "production" else None
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
)
app.include_router(inventory.router)
app.include_router(reports.router)
# app/config.py
from pydantic_settings import BaseSettings
from typing import List
class Settings(BaseSettings):
APP_NAME: str = "Warehouse Management API"
APP_VERSION: str = "2.1.0"
ENV: str = "development"
DATABASE_URL: str = "postgresql+asyncpg://localhost:5432/warehouse"
ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
Quick Start Guide
- Initialize the project: Run
uv init warehouse-api && cd warehouse-api to create a modern Python environment with uv or pip.
- Install dependencies: Execute
uv pip install fastapi uvicorn pydantic sqlalchemy asyncpg (or use the pyproject.toml above).
- Create the entry point: Add
main.py with the FastAPI instance, CORS middleware, and router includes from the configuration template.
- Launch the server: Run
uvicorn app.main:app --reload --port 8000 to start the development server with hot-reload enabled.
- Verify documentation: Navigate to
http://localhost:8000/api/docs to interact with the auto-generated Swagger UI and test endpoints without writing a single line of documentation code.