Lately, I’ve been thinking a lot about how to build systems that are not only powerful but also resilient and easy to grow. A question kept coming back to me: how do you connect independent services so that the whole system remains robust even when one part has a hiccup? This led me to combine some of my favorite Python tools—FastAPI, SQLAlchemy, and Redis—to create a setup that reacts to events. I’d like to show you how I approach this, sharing the patterns that have worked for me in production. If you stick around, I think you’ll find some useful ideas for your own projects.

Let’s talk about why this method is useful. In a typical setup, one service directly calls another. If the second service is slow or fails, the first one gets stuck waiting. An event-driven design changes this. Services communicate by sending and listening for notifications, called events, rather than calling each other directly. This means services can work independently. One can process a task now, while another picks it up later. The system becomes more flexible and can handle higher loads because work isn’t blocked.

You might be asking, where do these events live? We need a reliable, fast place to store and pass them along. This is where Redis Streams comes in. Think of it as a persistent, ordered log. A service can add an event to the log, and other services can read from it at their own pace. It’s a simple concept, but it’s the glue that holds everything together. Let’s look at how we set up a shared way to define these events.

We start by defining what an event looks like. Using Pydantic models gives us automatic validation and serialization.

from pydantic import BaseModel
from datetime import datetime
import uuid

class DomainEvent(BaseModel):
    event_id: str = str(uuid.uuid4())
    event_type: str
    aggregate_id: str
    occurred_at: datetime = datetime.utcnow()

    class Config:
        json_encoders = {datetime: lambda v: v.isoformat()}

class OrderCreated(DomainEvent):
    event_type: str = "OrderCreated"
    customer_id: str
    total_amount: float

Notice how each event gets a unique ID and a timestamp. This is crucial for tracking what happened and when. Now, how do we get these events into Redis? We create a dedicated client to handle the communication.

Here’s a basic producer that adds an event to a stream. We use the xadd command, which is Redis’s way of appending to a stream.

import redis.asyncio as redis
import json

async def publish_event(stream_name: str, event: DomainEvent):
    redis_client = redis.from_url("redis://localhost:6379")
    event_data = event.json()
    await redis_client.xadd(stream_name, {"payload": event_data})

On the other side, a consumer service needs to listen for new events. It reads from the stream, processes the data, and then acknowledges it so it won’t be read again.

async def consume_events(stream_name: str, consumer_group: str):
    redis_client = redis.from_url("redis://localhost:6379")
    while True:
        response = await redis_client.xreadgroup(
            groupname=consumer_group,
            consumername="inventory-service",
            streams={stream_name: '>'},
            count=1
        )
        for _, messages in response:
            for message_id, data in messages:
                event_payload = json.loads(data["payload"])
                # Process the event here...
                await redis_client.xack(stream_name, consumer_group, message_id)

With our messaging layer ready, we need a service to create these events. This is where FastAPI shines. We can build an API endpoint that, when hit, creates a record in its own database and publishes an event about it. This separates the immediate response to the user from the downstream work that needs to happen.

Imagine an order service. When a POST request creates an order, we save it using SQLAlchemy and then send an “OrderCreated” event.

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas, events

app = FastAPI()

@app.post("/orders", response_model=schemas.Order)
async def create_order(
    order_data: schemas.OrderCreate,
    db: AsyncSession = Depends(get_db)
):
    new_order = models.Order(**order_data.dict())
    db.add(new_order)
    await db.commit()
    
    event = events.OrderCreated(
        aggregate_id=str(new_order.id),
        customer_id=order_data.customer_id,
        total_amount=order_data.total_amount
    )
    await publish_event("order-stream", event)
    
    return new_order

But what happens if the database save works but the event publish fails? Or vice versa? This is a classic problem. We need to make sure these two steps are treated as a single unit of work, a transaction. One strategy is the Outbox Pattern. Instead of publishing directly, you first save the event to an outbox table in the same database transaction. Then, a separate process reads from this table and publishes to Redis. This guarantees the event is stored if the order is.

So, we have events flowing. But what about errors? If a consumer fails to process an event, we shouldn’t just lose it. A common practice is to use a Dead Letter Queue (DLQ). When processing fails after several retries, the event is moved to a separate stream for manual inspection. This prevents one bad event from blocking all others and gives you a clear place to debug.

async def process_with_dlq(event_data, max_retries=3):
    for attempt in range(max_retries):
        try:
            await handle_event(event_data)
            break
        except Exception as e:
            if attempt == max_retries - 1:
                await move_to_dlq(event_data, reason=str(e))

When you run this in production, seeing what’s happening is key. Structured logging with structlog and adding trace IDs to your events can help you follow a single request as it travels through multiple services. This observability turns a confusing distributed system into something you can actually debug.

Deploying this involves thinking about scale. You might run multiple instances of a consumer service. Redis Streams’ consumer groups are perfect here—they allow the work to be shared among instances, so you can process more events in parallel. Containerizing each service with Docker and orchestrating them with Kubernetes or using a managed cloud service gives you the control to scale up and down based on load.

Does this seem like a lot of moving parts? It can be. The payoff is a system where services don’t crash each other, where you can add new functionality just by listening to existing events, and where scaling is a matter of adding more containers. The initial setup is an investment that pays off in maintainability.

I’ve found this combination of FastAPI for clean APIs, SQLAlchemy for solid data handling, and Redis Streams for reliable messaging to be incredibly effective. It turns a collection of services into a coordinated system that can handle real-world demands. I hope walking through my approach gives you a practical starting point. What challenges have you faced when connecting microservices? Share your thoughts in the comments below—let’s learn from each other. If this guide was helpful, please like and share it with others who might be on a similar build journey.