Event-Driven Architecture — Event Sourcing, CQRS, and Pub/Sub with Kafka Examples

import json, datetime

# Event store — append-only log
event_store = []

def apply_event(event: dict):
    event["timestamp"] = datetime.datetime.utcnow().isoformat()
    event_store.append(event)

def get_account_balance(account_id: str) -> int:
    balance = 0
    for event in event_store:
        if event.get("account_id") == account_id:
            if event["type"] == "account_created":
                balance = event["initial_balance"]
            elif event["type"] == "money_deposited":
                balance += event["amount"]
            elif event["type"] == "money_withdrawn":
                balance -= event["amount"]
    return balance

# Simulate account operations
apply_event({"type": "account_created", "account_id": "ACC-001", "initial_balance": 1000})
apply_event({"type": "money_deposited", "account_id": "ACC-001", "amount": 500})
apply_event({"type": "money_withdrawn", "account_id": "ACC-001", "amount": 200})

print(f"Balance: {get_account_balance('ACC-001')}")
print(f"Event log:")
for e in event_store:
    print(f"  {e['timestamp']}: {e['type']} ${e.get('amount', e.get('initial_balance', ''))}")

# Command side (write)
class OrderCommandHandler:
    def __init__(self, event_store):
        self.event_store = event_store

    def place_order(self, user_id: str, items: list, total: float):
        event = {
            "type": "order_placed",
            "order_id": f"ORD-{hash(str(items))}",
            "user_id": user_id,
            "items": items,
            "total": total,
        }
        self.event_store.append(event)
        return event["order_id"]

# Query side (read)
class OrderQueryHandler:
    def __init__(self, read_db):
        self.read_db = read_db  # Materialized view

    def get_order_summary(self, user_id: str):
        return self.read_db.query("SELECT * FROM order_summary WHERE user_id = ?", user_id)

# Projection — builds the read model from events
class OrderProjection:
    def __init__(self, read_db):
        self.read_db = read_db

    def handle(self, event: dict):
        if event["type"] == "order_placed":
            self.read_db.execute(
                "INSERT INTO order_summary VALUES (?, ?, ?, ?)",
                event["order_id"], event["user_id"], event["total"], "pending"
            )

# producer.py — order service publishes events
from kafka import KafkaProducer
import json, time

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

order = {
    "order_id": "ORD-12345",
    "user_id": "USR-42",
    "items": [{"product": "widget", "qty": 2, "price": 9.99}],
    "total": 19.98,
    "timestamp": time.time()
}

producer.send('orders', value=order)
producer.flush()
print(f"Published order: {order['order_id']}")

# consumer.py — notification service reacts to events
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
    auto_offset_reset='earliest',
    group_id='notification-service'
)

print("Listening for orders...")
for message in consumer:
    order = message.value
    print(f"New order: {order['order_id']} by {order['user_id']}")
    print(f"Sending email confirmation for ${order['total']:.2f}")
    # In production: call email API

# Simple in-memory event bus
class EventBus:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, event_type: str, handler):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)

    def publish(self, event: dict):
        event_type = event["type"]
        for handler in self.subscribers.get(event_type, []):
            handler(event)

bus = EventBus()

# Services subscribe to events they care about
bus.subscribe("order_placed", lambda e: print(f"Inventory: reserve items for {e['order_id']}"))
bus.subscribe("order_placed", lambda e: print(f"Payment: charge {e['total']} for {e['order_id']}"))
bus.subscribe("order_placed", lambda e: print(f"Notification: email confirmation for {e['order_id']}"))
bus.subscribe("payment_received", lambda e: print(f"Shipping: prepare shipment for {e['order_id']}"))

# Place an order — all subscribers react
bus.publish({"type": "order_placed", "order_id": "ORD-123", "total": 29.99})

import time, uuid

class EventStore:
    def __init__(self):
        self.events = []

    def append(self, event):
        event["event_id"] = uuid.uuid4().hex[:8]
        event["timestamp"] = time.time()
        self.events.append(event)
        return event

class OrderService:
    def __init__(self, event_store: EventStore):
        self.store = event_store

    def place_order(self, user_id: str, product: str, qty: int, price: float):
        event = self.store.append({
            "type": "OrderPlaced",
            "data": {"user_id": user_id, "product": product, "qty": qty, "total": qty * price},
        })
        print(f"Order placed: {event['event_id']}")
        return event

class InventoryService:
    def handle(self, event):
        if event["type"] == "OrderPlaced":
            print(f"  Inventory: reserving {event['data']['qty']}x {event['data']['product']}")

class NotificationService:
    def handle(self, event):
        if event["type"] == "OrderPlaced":
            print(f"  Notification: email to {event['data']['user_id']}")

# Wire up the event-driven system
store = EventStore()
order_svc = OrderService(store)
inventory_svc = InventoryService()
notification_svc = NotificationService()

# Place an order — all services react via the event store
event = order_svc.place_order("USR-42", "Widget", 3, 9.99)

# Broadcast to all interested services
for service in [inventory_svc, notification_svc]:
    service.handle(event)