Lately, I’ve been thinking about how modern applications can stay responsive as they grow. The old way of services talking directly to each other often creates a fragile web of dependencies. One slow service can bring everything to a halt. This got me exploring a different path, one where services communicate through events, leading to systems that are more resilient and easier to scale. Today, I’ll walk you through building this using FastAPI, Redis Streams, and SQLAlchemy. Let’s build something that can handle growth gracefully.

Think of an event as a simple message that says, “Something happened.” In our case, an event could be “Order Created” or “Inventory Updated.” Services don’t send requests to each other. Instead, when something important occurs, they publish an event. Other services that care about that event can listen for it and act. This separation is powerful. It means you can update or add a new service without disrupting the ones already running.

What tool do we use to pass these messages around? We need something fast and reliable. This is where Redis Streams comes in. It’s not just a simple message queue. Redis Streams stores our events in an ordered log. Each event gets a unique ID, guaranteeing that every service processes events in the exact same order they happened. This is crucial for maintaining data consistency across your whole system.

Let’s look at how we define an event. We use Pydantic models to ensure our data is valid from the start.

from pydantic import BaseModel
import uuid
from datetime import datetime

class OrderCreatedEvent(BaseModel):
    event_id: str = str(uuid.uuid4())
    event_type: str = "OrderCreatedEvent"
    timestamp: datetime = datetime.utcnow()
    order_id: str
    customer_id: str
    items: list

Next, we need a bus, a central hub for our events. The EventBus class will handle talking to Redis Streams. Its main jobs are to publish events and allow services to subscribe to them.

import json
from redis.asyncio import Redis

class EventBus:
    def __init__(self, redis_client: Redis):
        self.redis = redis_client

    async def publish(self, stream_name: str, event: BaseModel):
        # Convert the event to a dictionary and publish it
        message = {"payload": json.dumps(event.dict())}
        await self.redis.xadd(stream_name, message)

Now, let’s build our first service with FastAPI. Imagine an order service. Its endpoint creates an order and then publishes an event. Notice how it doesn’t call the inventory service directly. It just announces the order to the world.

from fastapi import FastAPI, status
app = FastAPI()
event_bus = EventBus(redis_client)

@app.post("/orders", status_code=status.HTTP_201_CREATED)
async def create_order(order_data: dict):
    # ... logic to save the order to a database ...
    new_order_id = "ORD_123"

    # Create and publish the event
    event = OrderCreatedEvent(order_id=new_order_id, customer_id=order_data['customer_id'], items=order_data['items'])
    await event_bus.publish("order_events", event)
    return {"order_id": new_order_id}

Here’s an interesting question: what happens to the inventory after an order is placed? With our setup, the inventory service works independently. It listens to the order_events stream. When it sees an OrderCreatedEvent, it checks stock and updates its own database. If stock is low, it might publish a new event, like InventoryLowEvent. A notification service could then listen for that event to alert the warehouse team. See how one action creates a chain reaction?

This pattern, where you store every state change as an immutable event, is powerful. You can always rebuild your application’s state by replaying these events. We use SQLAlchemy to store these events in a database table, creating a complete and reliable history.

from sqlalchemy.orm import declarative_base
Base = declarative_base()

class EventStore(Base):
    __tablename__ = 'event_store'
    id = Column(Integer, primary_key=True)
    event_id = Column(String)
    event_type = Column(String)
    payload = Column(JSON)
    timestamp = Column(DateTime)

Of course, things don’t always go smoothly. A service might crash while processing an event. Redis Streams helps here with consumer groups. They track what each service has read. If a service restarts, it can pick up right where it left off. For events that repeatedly fail, we can move them to a “dead letter” stream for manual review.

Testing this setup requires a shift in thinking. You’re not just testing API endpoints. You need to verify that publishing an event triggers the correct reactions in other services. Tools like pytest and test containers are great for spinning up a temporary Redis instance to run your event flows in isolation.

The beauty of this approach is its clarity in operation. Because every action is an event, you have a clear trail of what happened and when. This makes debugging and monitoring much simpler. You can add metrics to track how many events are processed and how long it takes.

Building systems this way has changed how I view application design. It encourages breaking down problems into small, independent pieces that communicate through a clear, defined channel. This isn’t just a theory; it’s a practical method for building software that can evolve without constant rewrites.

I hope this guide gives you a solid starting point for your own event-driven projects. What kind of system are you thinking of building with these patterns? I’d love to hear about your ideas or answer any questions. If you found this useful, please share it with others who might be on a similar journey. Let’s keep the conversation going in the comments below.