I’ve been building microservices for years, and let me tell you – doing it right takes more than just coding skills. Last month, I saw another team launch a service that buckled under real traffic. That’s why I’m sharing this complete guide to production-ready microservices using FastAPI, SQLAlchemy, and Docker. Stick with me, and you’ll build systems that scale. Ready to ship robust services? Let’s go.

Our e-commerce system needs two core services: users and orders. Each runs independently but communicates when needed. Here’s how we structure it:

# Project structure
microservices-ecommerce/
├── user-service/
│   ├── app/
│   │   ├── database/
│   │   ├── models/
│   │   ├── routers/
│   ├── Dockerfile
├── order-service/
│   └── [similar]
├── docker-compose.yml

Start with dependencies. This requirements.txt covers essentials:

# requirements.txt
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
asyncpg==0.29.0  # Async PostgreSQL driver
pydantic==2.5.0   # Data validation
python-jose[cryptography]==3.3.0  # JWT tokens

For the User Service, database setup is critical. Notice how we use async sessions:

# user-service/app/database/connection.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

DATABASE_URL = "postgresql+asyncpg://user:password@db/ecommerce"

engine = create_async_engine(DATABASE_URL)
async_session = async_sessionmaker(engine, class_=AsyncSession)

async def get_db():
    async with async_session() as session:
        yield session

Ever wonder why UUIDs beat integers for public IDs? They prevent ID guessing attacks. Here’s our user model:

# user-service/app/models/user.py
import uuid
from sqlalchemy import Column, String, Boolean

class User(Base):
    __tablename__ = "users"
    public_id = Column(String(36), unique=True, default=lambda: str(uuid.uuid4()))
    email = Column(String(255), unique=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)

Pydantic keeps our data clean. This schema validates user creation:

# user-service/app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(..., min_length=8)
    confirm_password: str
    
    @field_validator('confirm_password')
    def passwords_match(cls, v, values):
        if v != values.data.get('password'):
            raise ValueError('Passwords must match')

Authentication is non-negotiable in production. We use JWT tokens with 30-minute expiry:

# shared/auth/jwt_handler.py
from jose import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your_strong_secret"
ALGORITHM = "HS256"

def create_access_token(data: dict):
    expire = datetime.utcnow() + timedelta(minutes=30)
    return jwt.encode({**data, "exp": expire}, SECRET_KEY, ALGORITHM)

Testing prevents midnight firefighting. This async test checks user registration:

# tests/test_users.py
from httpx import AsyncClient

async def test_create_user(async_client: AsyncClient):
    response = await async_client.post("/users/", json={
        "email": "test@example.com",
        "password": "securePass123",
        "confirm_password": "securePass123"
    })
    assert response.status_code == 201
    assert "public_id" in response.json()

Docker containers make deployment consistent. Here’s a minimal Dockerfile:

# user-service/Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

When services need to communicate, use Redis queues. But ask yourself: is this operation truly async? For order processing, yes. For auth checks? Probably not.

# order-service/app/services/notifications.py
import redis

r = redis.Redis(host='redis', port=6379)

def send_welcome_email(user_id):
    r.rpush('email_queue', f'welcome:{user_id}')

In production, health checks keep systems alive. Add this endpoint:

# In your main.py
from fastapi import APIRouter

router = APIRouter()

@router.get("/health")
async def health_check():
    return {"status": "OK", "timestamp": datetime.utcnow()}

I’ve deployed this pattern to Kubernetes clusters handling 10K+ RPM. The keys? Async I/O, proper connection pooling, and circuit breakers for inter-service calls. One mistake I made early on? Not setting SQLAlchemy’s pool size correctly. Too small, and requests starve. Too large, and you drown your database.

What separates hobby projects from production systems? Monitoring. Always track:

1. Error rates
2. P99 response times
3. Database connection wait times
4. Queue backpressure

I’ve walked you through the critical path – from code to containers. These patterns have saved my teams countless outages. If you implement just three things, make it async database access, proper JWT validation, and health checks.

What challenges have you faced with microservices? Share your stories below. If this guide helped you, pass it to a teammate – production readiness is a team sport. Comments? Questions? I’m here to help.