I’ve spent years building APIs that started simple but grew into tangled messes. Late nights debugging race conditions, performance bottlenecks that appeared out of nowhere, and deployment headaches made me wonder: there must be a better way. That’s when I found the combination of FastAPI, SQLAlchemy, and Docker. This isn’t just another tutorial—it’s the blueprint I wish I had when building systems that need to handle real traffic.
Let me show you how to build something that actually works in production.
First, our foundation. Why choose FastAPI over other frameworks? The answer lives in its design. FastAPI handles asynchronous operations naturally, which means your service can manage multiple requests without getting stuck waiting. Think of it like a restaurant with one waiter who can serve multiple tables simultaneously, rather than finishing one order before taking another. But how do we set this up properly?
from fastapi import FastAPI
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: create database connections
await database.connect()
yield
# Shutdown: clean up resources
await database.disconnect()
app = FastAPI(lifespan=lifespan)This lifespan manager ensures we handle connections correctly. Ever wondered why some APIs leak database connections? This pattern solves that.
Now, let’s talk data. SQLAlchemy 2.0 changed everything with native async support. Here’s how I structure models for an order system:
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
import uuid
class Base(DeclarativeBase):
pass
class Order(Base):
__tablename__ = "orders"
id: Mapped[str] = mapped_column(primary_key=True, default=lambda: str(uuid.uuid4()))
user_id: Mapped[int] = mapped_column(index=True)
total: Mapped[float]
created: Mapped[datetime] = mapped_column(default=datetime.utcnow)
# This relationship pattern prevents common N+1 query problems
items: Mapped[list["OrderItem"]] = relationship(back_populates="order")Notice the UUID primary key? This helps when you eventually need to merge data from multiple services. Integer IDs can collide across databases.
But what happens when things go wrong? Most tutorials show happy paths. Real services need solid error handling:
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
import logging
logger = logging.getLogger(__name__)
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
logger.warning(f"Business rule violation: {exc}")
return JSONResponse(
status_code=400,
content={"detail": str(exc), "type": "business_error"}
)
# In your route:
async def update_order(order_id: str):
order = await get_order(order_id)
if not order:
raise HTTPException(status_code=404, detail="Order not found")
if order.status == "shipped":
raise ValueError("Cannot modify shipped orders") # This gets handled properlyThis approach separates “not found” from “business rule violation.” Your frontend developers will thank you.
Security matters. Here’s a practical authentication setup:
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from datetime import datetime, timedelta
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
):
try:
payload = jwt.decode(
credentials.credentials,
SECRET_KEY,
algorithms=[ALGORITHM]
)
user_id: str = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid credentials")
return user_id
except JWTError:
raise HTTPException(status_code=401, detail="Could not validate credentials")
@app.get("/orders/")
async def list_orders(user_id: str = Depends(get_current_user)):
return await get_user_orders(user_id)This dependency injection pattern keeps security logic separate from business logic. Have you considered what happens when tokens expire? The HTTPBearer scheme handles the standard “Bearer” token format automatically.
Testing async code requires different thinking. Here’s my approach:
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
@pytest.mark.asyncio
async def test_create_order(async_client: AsyncClient, db_session: AsyncSession):
# Start a transaction that will roll back
async with db_session.begin():
response = await async_client.post(
"/orders/",
json={"user_id": 1, "items": [{"product_id": "p1", "quantity": 2}]}
)
assert response.status_code == 201
data = response.json()
assert "id" in data
# Verify in database
order = await db_session.get(Order, data["id"])
assert order.user_id == 1
# Transaction rolls back, test database stays cleanThis pattern ensures tests don’t interfere with each other. Each test gets a clean database state.
Now, the container. Docker makes deployment consistent:
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN useradd -m -u 1000 appuser
USER appuser
# Health check tells Docker when our service is ready
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD python -c "import httpx; httpx.get('http://localhost:8000/health', timeout=1)"
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]Notice the health check? It’s crucial for orchestration tools like Kubernetes to know if your service is healthy.
Monitoring often gets overlooked until midnight alerts wake you up. Here’s a simple but effective start:
from fastapi import Request
import time
from prometheus_client import Counter, Histogram
REQUEST_COUNT = Counter("http_requests_total", "Total HTTP requests")
REQUEST_LATENCY = Histogram("http_request_duration_seconds", "HTTP request latency")
@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
start_time = time.time()
REQUEST_COUNT.inc()
response = await call_next(request)
latency = time.time() - start_time
REQUEST_LATENCY.observe(latency)
return response
@app.get("/metrics")
async def metrics():
from prometheus_client import generate_latest
return Response(generate_latest())This gives you request counts and latency histograms. Combine with Grafana, and you’ll see problems before users complain.
Performance optimization comes last for a reason. Premature optimization wastes time. But when you need it, connection pooling and statement caching make a real difference:
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
pool_size=20, # Adjust based on your database limits
max_overflow=10,
pool_recycle=3600, # Recycle connections hourly
echo=False # Set to True only during development
)The pool_size should match your expected concurrent requests. Too small creates bottlenecks; too large overloads your database.
So, what’s the most common mistake I see? Mixing sync and async code without understanding the consequences. Async functions can call sync functions, but it blocks the entire event loop. Sync functions cannot call async functions directly. This boundary requires careful design.
I’ve walked you through the complete picture: from setup to monitoring. Each piece connects to others. The database models inform the API structure. The error handling supports the testing strategy. The container setup enables the monitoring.
This approach has served me well across multiple production systems. It scales from small projects to large distributed systems. The principles remain the same even as requirements grow.
What challenges have you faced with microservices? I’d love to hear about your experiences. If this guide helped you, consider sharing it with others who might benefit. Your comments and questions help improve this content for everyone. Let’s build more reliable systems together.
