Request-response is the default architecture for web applications, and for good reason. A client sends a request, a server processes it synchronously, and returns a response. Simple, predictable, easy to debug. This works beautifully for straightforward CRUD operations where a single action has a single effect.
It starts to break when a single user action triggers multiple downstream effects. Consider what happens when a user places an order on an e-commerce platform. The system needs to: validate and process the payment, reserve inventory for the ordered items, send a confirmation email to the customer, notify the warehouse to begin fulfillment, update the real-time analytics dashboard, calculate and award loyalty points, run a fraud detection check, and potentially notify a Slack channel that the sales team monitors. In a request-response architecture, all of these happen synchronously inside the order handler. The customer waits for all of them to complete before seeing a confirmation page. The order handler grows into a 400-line function that imports modules for payments, email, inventory, analytics, loyalty programs, fraud detection, and messaging. Adding a new downstream action (say, sending a webhook to a partner system) means modifying the order handler, testing the change, and deploying it — even though the core ordering logic has not changed.
Event-driven architecture (EDA) solves this coupling problem by separating the announcement of “something happened” from the reactions to that announcement. The order handler publishes an “order placed” event containing all relevant details, and returns immediately with a confirmation. Independent consumers — each owned by a different team, each deployed independently — handle the email, inventory update, analytics recording, and everything else asynchronously. Adding a new reaction means deploying a new consumer. Zero changes to the order handler or any existing consumer.
Here is how to build production-grade event-driven systems in Python, covering event design, message broker selection, consumer patterns, error handling, and the operational reality of running EDA at scale.
Core Concepts
Event-driven architecture has three fundamental components that interact through a message broker:
- Events are immutable records of something that happened. “Order #456 was placed by customer C-789 for $142.50.” “Payment of $142.50 was successfully processed via Stripe.” “Inventory for SKU-123 dropped below the reorder threshold of 10 units.” Events are facts about the past, not instructions for the future. This distinction matters: events describe what happened, not what should happen.
- Producers are services that detect or cause events and publish them to the message broker. The order service produces “order placed” events. The payment service produces “payment processed” events. The inventory service produces “inventory low” events. A producer publishes the event and immediately moves on — it does not know or care which consumers will react to it.
- Consumers are services that subscribe to specific event types and react to them. The email service consumes “order placed” events and sends confirmation emails. The analytics service consumes all events and updates dashboards. The fraud service consumes “order placed” events and runs risk scoring. Each consumer has a single, focused responsibility.
The message broker sits between producers and consumers, providing three crucial capabilities: durability (events are not lost if consumers are temporarily unavailable), ordering (events from the same source arrive in the order they were produced), and delivery guarantees (at-least-once delivery ensures every consumer eventually processes every relevant event).
Designing Events Well
Event design is the single most important architectural decision in an EDA system and the one most commonly botched by teams new to the pattern. A well-designed event is self-contained, versioned, and carries enough context for any reasonable consumer to act on it without making synchronous calls to other services.
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional, Any
import uuid
import json
@dataclass
class Event:
"""Base class for all domain events."""
event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
event_type: str = '' # Set by each concrete event class
event_version: int = 1 # Increment when schema changes
timestamp: str = field(
default_factory=lambda: datetime.utcnow().isoformat() + 'Z'
)
source: str = '' # Which service produced this
correlation_id: Optional[str] = None # Trace across service boundaries
causation_id: Optional[str] = None # Which event caused this one
def to_json(self) -> str:
return json.dumps(asdict(self), default=str, ensure_ascii=False)
@classmethod
def from_json(cls, raw: str) -> 'Event':
data = json.loads(raw)
return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
@dataclass
class OrderPlaced(Event):
event_type: str = 'order.placed'
source: str = 'order-service'
# Include everything a consumer might need
order_id: str = ''
customer_id: str = ''
customer_email: str = ''
customer_name: str = ''
items: list = field(default_factory=list)
subtotal: float = 0.0
tax: float = 0.0
shipping: float = 0.0
total: float = 0.0
currency: str = 'USD'
shipping_address: dict = field(default_factory=dict)
billing_address: dict = field(default_factory=dict)
payment_method: str = '' # 'credit_card', 'paypal', etc.
@dataclass
class PaymentProcessed(Event):
event_type: str = 'payment.processed'
source: str = 'payment-service'
payment_id: str = ''
order_id: str = ''
amount: float = 0.0
currency: str = 'USD'
method: str = '' # 'credit_card', 'paypal', 'bank_transfer'
processor: str = '' # 'stripe', 'braintree'
status: str = '' # 'success', 'failed', 'pending'
failure_reason: Optional[str] = None
@dataclass
class InventoryLow(Event):
event_type: str = 'inventory.low'
source: str = 'inventory-service'
sku: str = ''
product_name: str = ''
current_quantity: int = 0
reorder_threshold: int = 0
reorder_quantity: int = 0
warehouse_id: str = ''
supplier_id: Optional[str] = NoneThe design rules that matter most in practice:
- Include all context a consumer needs to act independently. The
OrderPlacedevent includes the customer email and name, not just the customer ID. This means the email consumer does not need to call the customer service to send a confirmation. This dramatically reduces coupling: if the customer service is having a bad day, emails still go out. Every consumer that needs customer data should find it in the event, not fetch it synchronously. - Version events from the start. When you inevitably need to add, rename, or restructure fields, increment the version number. Consumers should handle multiple versions gracefully: accept unknown fields (forward compatibility), provide sensible defaults for missing fields (backward compatibility). Never remove a field — deprecate it by documenting that it will always be null/empty in version N+.
- Include correlation and causation IDs. When an order placement triggers a payment, which triggers a fulfillment request, which triggers a shipping notification, the correlation ID traces the entire chain back to the original user action. The causation ID tracks the immediate parent: the fulfillment event was caused by the payment event, which was caused by the order event. Together, these IDs enable distributed tracing that makes debugging event chains feasible.
- Events are always past tense.
OrderPlaced, notPlaceOrder.PaymentProcessed, notProcessPayment. Events describe facts about the past. Commands (imperative instructions like “process this payment”) are a different pattern with different semantics and different reliability requirements.
Choosing a Message Broker
The message broker is the infrastructure backbone of your EDA system. The choice depends on throughput requirements, durability needs, operational complexity tolerance, and your team’s existing expertise:
Redis Streams (our default choice for internal systems): Good for systems processing up to approximately 100,000 events per second. Consumer groups provide at-least-once delivery with built-in acknowledgment tracking. Data persists in memory with optional AOF disk persistence. We use Redis Streams when we already run Redis for caching and queues because it avoids introducing another infrastructure component.
RabbitMQ: Mature, well-documented, excellent routing capabilities. Supports multiple exchange patterns (fanout for broadcast, topic for pattern-matched routing, direct for point-to-point). Better durability guarantees than Redis Streams. Good for systems needing sophisticated message routing. Throughput ceiling of 50-80K messages per second on a single node.
Apache Kafka: The industry standard for high-throughput, durable event streaming. Capable of millions of events per second. Persistent, append-only log that can be replayed from any point in history. Consumer groups with partition-based parallelism. Operationally complex — requires ZooKeeper (or KRaft in newer versions), careful partition planning, and monitoring for consumer lag. Do not use Kafka unless you genuinely need its scale or its log replay capability.
PostgreSQL LISTEN/NOTIFY: If you already run PostgreSQL and your event volume is under 1,000 events per second, this is the simplest possible option: zero additional infrastructure. Notifications are ephemeral (lost if no listener is connected), so this works only for supplementary real-time updates, not for critical event processing.
Here is a production event bus implementation using Redis Streams:
import redis
import json
import time
from typing import Callable
import logging
logger = logging.getLogger(__name__)
class EventBus:
def __init__(self, redis_client: redis.Redis, max_stream_length: int = 100000):
self.r = redis_client
self.max_len = max_stream_length
def publish(self, event: Event, stream: str = None):
"""Publish an event to a Redis Stream."""
stream_name = stream or f"events:{event.event_type.split('.')[0]}"
message_id = self.r.xadd(stream_name, {
'event_type': event.event_type,
'event_id': event.event_id,
'data': event.to_json()
}, maxlen=self.max_len)
logger.info(
f"Published {event.event_type} ({event.event_id}) "
f"to {stream_name} as {message_id}"
)
return message_id
def subscribe(
self, stream: str, group: str, consumer_name: str,
handler: Callable, batch_size: int = 10, block_ms: int = 5000
):
"""Subscribe to events with a consumer group."""
# Create consumer group idempotently
try:
self.r.xgroup_create(stream, group, id='0', mkstream=True)
except redis.ResponseError as e:
if 'BUSYGROUP' not in str(e):
raise
logger.info(
f"Consumer {consumer_name} subscribed to {stream} "
f"in group {group}"
)
while True:
try:
messages = self.r.xreadgroup(
group, consumer_name,
{stream: '>'},
count=batch_size,
block=block_ms
)
if not messages:
continue
for stream_name, entries in messages:
for message_id, data in entries:
event_type = data.get(b'event_type', b'').decode()
try:
event_data = json.loads(data[b'data'])
handler(event_data)
self.r.xack(stream, group, message_id)
except Exception as e:
logger.error(
f"Error processing {message_id} "
f"({event_type}): {e}",
exc_info=True
)
# Don't ack -- will be redelivered
except redis.ConnectionError:
logger.warning("Redis connection lost, reconnecting...")
time.sleep(1)
except Exception as e:
logger.error(f"Unexpected error in consumer loop: {e}", exc_info=True)
time.sleep(1)Consumer Patterns
Consumers are the workers that react to events. Each consumer should have a single, well-defined responsibility and be independently deployable. Here are the three most common patterns we use:
Pattern 1: Simple Event Handler
The simplest pattern: receive an event, perform one action, acknowledge. Stateless and straightforward.
class EmailNotificationConsumer:
"""Sends confirmation emails when orders are placed."""
def __init__(self, email_service, template_engine):
self.email = email_service
self.templates = template_engine
def handle(self, event_data: dict):
if event_data['event_type'] != 'order.placed':
return # Ignore events we don't care about
html = self.templates.render('order_confirmation', {
'customer_name': event_data['customer_name'],
'order_id': event_data['order_id'],
'items': event_data['items'],
'total': event_data['total'],
'currency': event_data['currency'],
'shipping_address': event_data['shipping_address']
})
self.email.send(
to=event_data['customer_email'],
subject=f"Order Confirmation - {event_data['order_id']}",
html=html
)
logger.info(f"Sent confirmation email for order {event_data['order_id']}")Pattern 2: Stateful Event Aggregator
Consumes events from multiple sources and maintains aggregated state — ideal for analytics, reporting dashboards, and materialized views.
class RealTimeAnalyticsConsumer:
"""Aggregates events into real-time metrics."""
def __init__(self, db):
self.db = db
def handle(self, event_data: dict):
event_type = event_data['event_type']
event_date = event_data.get('timestamp', '')[:10] # YYYY-MM-DD
if event_type == 'order.placed':
self._increment_metric(event_date, 'orders_count', 1)
self._increment_metric(event_date, 'revenue_total', event_data['total'])
self._increment_metric(
event_date, f"revenue_{event_data['currency'].lower()}",
event_data['total']
)
elif event_type == 'payment.processed':
status = event_data['status']
self._increment_metric(event_date, f"payments_{status}", 1)
if status == 'failed':
self._increment_metric(
event_date, 'payment_failures_amount', event_data['amount']
)
elif event_type == 'inventory.low':
self.db.execute("""
INSERT INTO alerts (type, entity_id, message, severity, created_at)
VALUES ('inventory_low', %s, %s, 'warning', NOW())
""", (
event_data['sku'],
f"{event_data['product_name']} at {event_data['current_quantity']} units "
f"in warehouse {event_data['warehouse_id']}"
))
def _increment_metric(self, date: str, metric: str, value: float):
self.db.execute("""
INSERT INTO daily_metrics (date, metric, value)
VALUES (%s, %s, %s)
ON CONFLICT (date, metric)
DO UPDATE SET value = daily_metrics.value + EXCLUDED.value
""", (date, metric, value))Pattern 3: Saga / Process Manager
Orchestrates multi-step business processes that span multiple services, tracking state and triggering compensating actions when steps fail.
class OrderFulfillmentSaga:
"""
Manages the order fulfillment workflow:
Order Placed -> Payment -> Inventory Reservation -> Shipment Scheduling
With compensating actions on failure at any step.
"""
def __init__(self, event_bus: EventBus, db):
self.bus = event_bus
self.db = db
def handle(self, event_data: dict):
event_type = event_data['event_type']
order_id = event_data.get('order_id')
correlation_id = event_data.get('correlation_id', event_data['event_id'])
if event_type == 'order.placed':
self._update_state(order_id, 'payment_pending')
# Trigger payment processing
self.bus.publish(RequestPayment(
order_id=order_id,
amount=event_data['total'],
currency=event_data['currency'],
payment_method=event_data['payment_method'],
correlation_id=correlation_id,
causation_id=event_data['event_id']
))
elif event_type == 'payment.processed':
if event_data['status'] == 'success':
self._update_state(order_id, 'inventory_reserving')
self.bus.publish(ReserveInventory(
order_id=order_id,
items=self._get_order_items(order_id),
correlation_id=correlation_id,
causation_id=event_data['event_id']
))
else:
# Payment failed -- cancel the order
self._update_state(order_id, 'cancelled')
self.bus.publish(OrderCancelled(
order_id=order_id,
reason=f"Payment failed: {event_data.get('failure_reason', 'unknown')}",
correlation_id=correlation_id
))
elif event_type == 'inventory.reserved':
self._update_state(order_id, 'shipping_scheduled')
self.bus.publish(ScheduleShipment(
order_id=order_id,
warehouse_id=event_data['warehouse_id'],
correlation_id=correlation_id
))
elif event_type == 'inventory.reservation_failed':
# Inventory unavailable -- refund payment and cancel
self._update_state(order_id, 'cancelling')
self.bus.publish(RefundPayment(
order_id=order_id,
reason='Items out of stock',
correlation_id=correlation_id
))The saga pattern is essential for business processes that span multiple services where partial completion is unacceptable. Each step is triggered by an event from the previous step. When a step fails, the saga executes compensating actions to undo the effects of completed steps — refunding the payment if inventory reservation fails, for example. Without a saga, you end up with orders stuck in inconsistent states: payment charged but items never shipped, inventory reserved but never released.
Error Handling and Dead Letter Queues
In event-driven systems, consumer errors are inevitable and must be handled gracefully. The standard approach combines retry logic with a dead-letter queue (DLQ) for permanently failing events:
class ResilientConsumer:
def __init__(self, handler: Callable, event_bus: EventBus, max_retries: int = 3):
self.handler = handler
self.bus = event_bus
self.max_retries = max_retries
def process(self, event_data: dict, message_id: str):
retries = event_data.get('_retry_count', 0)
try:
self.handler(event_data)
except TransientError as e:
# Retryable: network timeout, temporary DB unavailability
if retries < self.max_retries:
event_data['_retry_count'] = retries + 1
event_data['_last_error'] = str(e)
delay = min(2 ** retries, 30) # Exponential backoff, capped
time.sleep(delay)
self.bus.republish(event_data)
else:
self._dead_letter(event_data, str(e), 'max_retries_exceeded')
except PermanentError as e:
# Non-retryable: invalid data, business rule violation
self._dead_letter(event_data, str(e), 'permanent_failure')
except Exception as e:
# Unknown: treat as transient for first few attempts
if retries < self.max_retries:
event_data['_retry_count'] = retries + 1
self.bus.republish(event_data)
else:
self._dead_letter(event_data, str(e), 'unknown_error')
def _dead_letter(self, event_data: dict, error: str, reason: str):
self.bus.publish_to_dlq({
'original_event': event_data,
'error': error,
'reason': reason,
'consumer': self.handler.__class__.__name__,
'dead_lettered_at': datetime.utcnow().isoformat(),
'retry_count': event_data.get('_retry_count', 0)
})
logger.error(
f"Dead-lettered event {event_data.get('event_id')}: {reason} - {error}"
)The DLQ is your canary in the coal mine. Monitor its depth daily. A growing DLQ means your system has problems that automated retries cannot solve: a breaking schema change, a downstream service that is permanently misconfigured, or a bug in consumer logic. Build a simple admin interface that lets operators inspect dead-lettered events, understand why they failed, fix the underlying issue, and replay them with a single click.
Testing Event-Driven Systems
Testing EDA systems requires thinking differently from request-response testing. You cannot just call an endpoint and check the response. You need to verify that the right events were published with the right data, and that consumers handle events correctly including edge cases and duplicates.
class InMemoryEventBus:
"""Test double that captures published events for assertion."""
def __init__(self):
self.published = []
self._handlers = {}
def publish(self, event: Event, stream: str = None):
self.published.append(event)
for handler in self._handlers.get(event.event_type, []):
handler(json.loads(event.to_json()))
def on(self, event_type: str, handler: Callable):
self._handlers.setdefault(event_type, []).append(handler)
def assert_published(self, event_type: str, count: int = 1, **match):
matching = [
e for e in self.published
if e.event_type == event_type
and all(getattr(e, k) == v for k, v in match.items())
]
assert len(matching) == count, (
f"Expected {count} '{event_type}' events matching {match}, "
f"found {len(matching)}. All published: "
f"{[e.event_type for e in self.published]}"
)
def assert_not_published(self, event_type: str):
matching = [e for e in self.published if e.event_type == event_type]
assert len(matching) == 0, (
f"Expected no '{event_type}' events, found {len(matching)}"
)
# Example test
def test_order_placement_publishes_event():
bus = InMemoryEventBus()
service = OrderService(event_bus=bus, db=mock_db)
service.place_order(
customer_id='cust-1',
items=[{'sku': 'WIDGET-A', 'qty': 2, 'price': 29.99}]
)
bus.assert_published('order.placed', count=1, customer_id='cust-1')
def test_email_consumer_handles_order_placed():
mock_email = MockEmailService()
consumer = EmailNotificationConsumer(email_service=mock_email)
consumer.handle({
'event_type': 'order.placed',
'event_id': 'evt-123',
'order_id': 'ord-456',
'customer_email': 'test@example.com',
'customer_name': 'Test User',
'items': [{'name': 'Widget A', 'qty': 2}],
'total': 59.98,
'currency': 'USD'
})
assert mock_email.sent_count == 1
assert mock_email.last_recipient == 'test@example.com'
def test_email_consumer_is_idempotent():
"""Processing the same event twice should not send duplicate emails."""
mock_email = MockEmailService()
consumer = EmailNotificationConsumer(email_service=mock_email)
event = {
'event_type': 'order.placed',
'event_id': 'evt-123', # Same event ID both times
'order_id': 'ord-456',
'customer_email': 'test@example.com',
'customer_name': 'Test User',
'items': [],
'total': 0,
'currency': 'USD'
}
consumer.handle(event)
consumer.handle(event) # Duplicate delivery
assert mock_email.sent_count == 1 # Should only send onceObservability: Seeing Through Distributed Asynchrony
Debugging an event-driven system without proper observability is debugging a distributed system in the dark. Since actions happen asynchronously across multiple services with no synchronous call chain, traditional request-scoped logging and tracing do not work. You need purpose-built observability:
- Distributed tracing with correlation IDs. Every event carries a correlation ID that traces back to the originating user action. Every consumer creates a trace span linked to this ID. OpenTelemetry is the standard. The result: a complete timeline showing how one user's order placement flowed through payment processing, inventory reservation, email notification, and analytics recording -- even though each happened in a different service, on a different machine, at a different time.
- Event flow dashboard. A real-time visualization of events per second by type, consumer lag (how far behind each consumer group is), processing error rates by consumer, and DLQ depth. This is your operational nerve center. Consumer lag is the most critical metric: if a consumer falls behind, events accumulate, and users experience growing delays in downstream effects (late emails, stale dashboards, delayed fulfillment).
- Consumer health metrics per consumer. Processing throughput (events per second), latency percentiles (p50, p95, p99 processing time), error rate, and queue depth. Alert when any consumer's lag exceeds a threshold or its error rate spikes. Each consumer should expose these as Prometheus metrics.
- Event replay capability. The ability to replay events from a specific point in time through a specific consumer. This is essential for two scenarios: recovering from a consumer bug that processed events incorrectly (fix the bug, replay the affected events), and bootstrapping a new consumer that needs to process historical events to build up its state.
Common Pitfalls to Avoid
Having built and operated several EDA systems over the past four years, these are the mistakes I see most frequently in teams adopting the pattern:
Events that are too fat. Including every conceivable detail -- the full customer profile, complete order history, related product recommendations -- in every event. Events should carry enough context for consumers to act, not a database dump. If a consumer needs the customer's full order history, it should maintain its own read model populated by consuming events over time, not expect the order history in every new event.
Events that are too thin. The opposite problem: events containing only an entity ID that forces every consumer to make a synchronous API call to fetch the actual data. "Order 456 was placed" with no details forces the email consumer to call the order service, the analytics consumer to call the order service, the fraud consumer to call the order service. You have not decoupled anything -- you have added a message broker between tightly coupled services.
Missing idempotency in consumers. At-least-once delivery means events will occasionally be delivered more than once (broker redelivery after a timeout, consumer restart before acknowledgment). Every consumer must handle duplicate events gracefully. The simplest approach: maintain a set of processed event IDs (in Redis or a database table) and skip events you have already processed. For consumers with external side effects (sending emails, calling APIs), this deduplication is critical.
Synchronous expectations in an async system. A producer publishes an event and then immediately queries the consumer's database expecting to find results. The consumer has not processed the event yet because it is asynchronous. Eventual consistency is inherent in EDA. Design your user-facing interfaces to accommodate this: show "Order confirmed, processing" immediately, and update to "Payment confirmed" and "Shipped" as those events flow through the system.
No schema registry or shared event definitions. As the system grows beyond two or three services, keeping event schemas consistent across all producers and consumers becomes critical. Use a shared library (a Python package containing all event dataclass definitions) that all services depend on. Publish it to your private package registry. When an event schema changes, the library version bump forces all consumers to acknowledge the change at upgrade time.
Conclusion
Event-driven architecture is not a universal solution. It adds genuine complexity: eventual consistency that requires different UI patterns, message ordering challenges for events that depend on each other, debugging difficulty when causal chains span five services and three minutes, and operational overhead of running a message broker reliably. For simple CRUD applications with straightforward request-response flows, EDA is overkill that adds cost without proportional benefit.
But for systems where a single action triggers multiple independent downstream reactions, where services need to evolve and deploy independently, where scalability requires decoupling the rate of production from the rate of consumption, and where resilience means a failing email service should not prevent orders from being placed -- for these systems, EDA is the right architecture. The patterns described here -- carefully designed events, a reliable message broker, resilient consumers with DLQ handling, saga orchestration for multi-step processes, and comprehensive distributed observability -- form a production-tested foundation for event-driven Python systems.
The best adoption strategy is incremental. Pick one workflow in your application that currently has a monolithic handler doing too many things. Extract the secondary actions (email, analytics, notifications) into event consumers. Run it alongside the existing synchronous code, verify it works, then remove the old calls. Expand to the next workflow. Event-driven architecture is best adopted one workflow at a time, not as a big-bang rewrite that risks the entire system on a pattern the team has never operated in production.
