Lately, I’ve been wrestling with scaling challenges in monolithic applications. As user bases grow, single codebases become unwieldy - deployments turn risky, databases choke under load, and adding features feels like navigating a maze. This struggle led me to explore microservices with Python tools that balance power with simplicity. Today, I’ll share a production-ready approach combining FastAPI, SQLAlchemy, and Redis that transformed how I build systems.

Our architecture features three independent services: Users handles authentication and profiles, Products manages inventory and search, and Orders processes transactions. Each runs in its own container, scales independently, and communicates via REST APIs or messaging queues. Why this separation? Imagine needing to update payment logic without affecting user profiles. Microservices make this possible.

Let’s start with dependencies. Create a base requirements.txt with these key packages:

fastapi==0.104.1
sqlalchemy==2.0.23
redis==5.0.1
httpx==0.25.2
tenacity==8.2.3

For database operations, I developed a reusable SQLAlchemy core. Notice connection pooling and session management:

# shared/database.py
class DatabaseManager:
    def __init__(self, database_url: str):
        self.engine = create_async_engine(
            database_url,
            pool_size=20,
            max_overflow=30,
            pool_pre_ping=True
        )
        self.async_session = async_sessionmaker(
            self.engine, expire_on_commit=False
        )
    
    @asynccontextmanager
    async def get_session(self):
        async with self.async_session() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise

This context manager handles transactions automatically. Now, what happens when services need to share data? Our User service demonstrates repository patterns:

# user-service/repositories.py
class UserRepository(BaseRepository):
    async def get_by_email(self, email: str) -> Optional[User]:
        result = await self.session.execute(
            select(User).where(User.email == email)
        return result.scalar_one_or_none()

Authentication uses JWT tokens. Notice how we verify tokens across services without shared state:

# shared/auth.py
def verify_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

For the Product service, Redis caching boosts performance. Our strategy: cache database queries but invalidate on writes. How much faster? In tests, response times dropped from 200ms to 8ms. Here’s the implementation:

# product-service/services.py
async def get_product(product_id: int):
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    product = await product_repo.get_by_id(product_id)
    await redis.setex(cache_key, 300, json.dumps(product.dict()))
    return product

Order processing introduces async communication. When creating an order, we deduct inventory via message queue:

# order-service/services.py
async def create_order(user_id: int, items: list):
    order = await order_repo.create(user_id=user_id)
    await message_queue.enqueue("inventory_update", 
                                payload={"order_id": order.id, "items": items})
    return order

But what if the Product service is down? Circuit breakers prevent cascading failures:

# shared/http_client.py
@circuit_breaker(failure_threshold=5, recovery_timeout=30)
async def post(url: str, data: dict):
    async with httpx.AsyncClient() as client:
        return await client.post(url, json=data, timeout=2.0)

For observability, Structlog with OpenTelemetry traces requests across services:

# shared/logger.py
structlog.configure(
    processors=[structlog.processors.JSONRenderer()],
    context_class=dict,
    wrapper_class=structlog.BoundLogger
)

Deployment uses Docker Compose. Notice how each service connects only to its dependencies:

# docker-compose.yml
services:
  user-service:
    build: ./user-service
    depends_on:
      - user-db
    environment:
      DB_URL: postgresql+asyncpg://user:pass@user-db:5432/users

  product-service:
    build: ./product-service
    depends_on:
      - product-db
      - redis

Common pitfalls? Service boundaries are critical. I once split by technical layers instead of business capabilities - a mistake causing constant cross-service changes. Another lesson: always version APIs from day one. Changing endpoints without versioning breaks dependent services during updates.

For testing, I mount test containers with sample data. Pytest fixtures spin up databases and services:

# tests/conftest.py
@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

After implementing this, our deployment frequency increased 5x while error rates dropped. The true win? Teams now deploy independently without coordination meetings. What could your team build with this foundation?

Try this approach on your next project. Share your experiences in the comments below - I’d love to hear what challenges you faced or improvements you discovered. If this helped you, consider sharing it with others facing similar architecture decisions.