I’ve spent the last three months migrating our legacy monolithic system to microservices. The complexity of maintaining intertwined services while ensuring scalability pushed me toward FastAPI - and what a revelation it’s been. Today I’ll share the blueprint we created for production-grade microservices that handle 50K+ RPM daily. Grab your favorite beverage and let’s build something robust together.

Setting up the foundation correctly saves countless debugging hours later. We structure projects like this:

user-service/
├── app/
│   ├── core/         # Configs, security, DB
│   ├── models/       # SQLAlchemy models
│   ├── schemas/      # Pydantic validators
│   ├── services/     # Business logic
│   ├── api/          # Endpoint routers
│   └── utils/        # Logging, helpers

Our dependency setup uses Poetry for pinning exact versions - critical for reproducibility. Notice how we separate production and dev dependencies:

[tool.poetry.dependencies]
fastapi = "0.104.1"
sqlalchemy = "2.0.23"
asyncpg = "0.29.0"
pydantic = { extras = ["email"], version = "2.5.0" }

[tool.poetry.group.dev.dependencies]
pytest = "7.4.3"
pytest-asyncio = "0.21.1"

The database layer deserves special attention. SQLAlchemy 2.0’s async support changes everything. Here’s our config pattern:

# app/core/config.py
class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://user:password@db/userdb"
    DATABASE_POOL_SIZE: int = 20
    MAX_OVERFLOW: int = 30
    SECRET_KEY: str = secrets.token_urlsafe(32)

Connection management is where many stumble. Notice how we handle sessions:

# app/core/database.py
engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=settings.DATABASE_POOL_SIZE,
    max_overflow=settings.DATABASE_MAX_OVERFLOW,
    pool_pre_ping=True
)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

For user models, we enforce security at the ORM level. How often do you see password hashing baked into the model itself?

# app/models/user.py
class User(Base):
    __tablename__ = "users"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String(255), unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    @validates("password")
    def hash_password(self, key, password):
        return get_password_hash(password)

Authentication is where many APIs fail. We implement JWT with refresh tokens:

# app/core/security.py
def create_access_token(data: dict) -> str:
    expires = datetime.utcnow() + timedelta(minutes=15)
    return jwt.encode({**data, "exp": expires}, SECRET_KEY, ALGORITHM)

def verify_password(plain_pwd, hashed_pwd) -> bool:
    return pwd_context.verify(plain_pwd, hashed_pwd)

Pydantic models prevent garbage-in/garbage-out scenarios. Notice the email validator:

# app/schemas/user.py
class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, pattern=r"^(?=.*[A-Za-z])(?=.*\d).*$")

    @model_validator(mode="after")
    def validate_password_strength(self):
        if len(self.password) < 12:
            raise ValueError("Password too weak")
        return self

For long-running tasks, we combine Celery with Redis. How do you handle week-long report generations without blocking your API?

# app/utils/tasks.py
@app.post("/reports")
async def create_report(background_tasks: BackgroundTasks):
    background_tasks.add_task(generate_user_report)
    return {"status": "processing"}

@celery.task
def generate_user_report():
    # Heavy processing here
    build_csv_export.delay()  # Chained tasks

Containerization solved our “works on my machine” nightmares. The Dockerfile optimizes layers:

# Dockerfile
FROM python:3.11-slim

RUN pip install poetry
COPY pyproject.toml poetry.lock ./
RUN poetry install --no-root --no-dev

COPY . .
CMD ["poetry", "run", "uvicorn", "app.main:app"]

Testing deserves equal attention to production code. We use this pattern:

# tests/test_users.py
async def test_user_flow(async_client):
    # Create
    response = await async_client.post("/users", json={"email": "test@example.com"})
    assert response.status_code == 201
    
    # Retrieve
    user_id = response.json()["id"]
    response = await async_client.get(f"/users/{user_id}")
    assert response.json()["email"] == "test@example.com"

In production, we learned the hard way: always implement structured logging. This setup saves hours during incidents:

# app/utils/logger.py
def setup_logging():
    structlog.configure(
        processors=[
            structlog.processors.JSONRenderer()
        ],
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory()
    )

Our monitoring stack includes Prometheus metrics and health checks:

# app/core/monitoring.py
@app.get("/health")
def health_check():
    return {"status": "ok", "services": {"db": db_online()}}

@app.get("/metrics")
def metrics():
    return generate_latest(REGISTRY)

Common pitfalls we encountered:

• Forgetting pool_pre_ping=True causing stale connections
• Not setting expire_on_commit=False in sessions
• Missing @model_validator checks in Pydantic models
• Underestimating connection pool requirements

Did you know that 70% of API failures originate from validation gaps? Our Pydantic setup catches malformed data before it touches business logic.

This architecture now serves millions of requests daily across 12 services. The true beauty? Adding new features takes hours, not days. What would you build with this foundation?

If this guide helped you, pay it forward - share with your team and leave a comment about your microservice journey. Your war stories help us all improve.