Event Sourcing

Event Sourcing im Überblick

Bei Event Sourcing speicherst du Ereignisse, die zu einem Zustand geführt haben. Der aktuelle Zustand wird durch Abspielen relevanter Events oder durch Kombination aus Snapshot und neuen Events rekonstruiert.

Traditionell (State-basiert):

Datenbank: { balance: current_balance }

→ Du weißt: aktueller Kontostand
→ Ohne zusätzliche Historie weißt du nicht vollständig, wie es dazu kam

Event Sourcing:

Event Store:
1. AccountCreated { balance: initial_balance }
2. MoneyDeposited { amount: deposit_amount }
3. MoneyWithdrawn { amount: withdrawal_amount }

→ Replay: Zustand aus Events rekonstruieren
→ Du siehst, welche Änderungen passiert sind

Vorteile:

Technisch betrachtet

Event Store Struktur

┌─────────────────────────────────────────────────────┐
│                    Event Store                       │
├──────────┬──────────┬───────────────────────────────┤
│ Event ID │ Stream   │ Event Data                    │
├──────────┼──────────┼───────────────────────────────┤
│ event-1  │ order-1  │ OrderCreated { items: [...] } │
│ event-2  │ order-1  │ ItemAdded { sku: "SKU_A" }    │
│ event-3  │ order-1  │ ItemRemoved { sku: "SKU_B" }  │
│ event-4  │ order-1  │ OrderSubmitted { total: ... } │
│ event-5  │ order-2  │ OrderCreated { items: [...] } │
└──────────┴──────────┴───────────────────────────────┘

Streams: Gruppieren Events zu einer Entität (z.B. eine Bestellung).

Event-Klassen

from dataclasses import dataclass
from datetime import datetime
from typing import List

@dataclass
class Event:
    event_id: str
    timestamp: datetime
    
@dataclass
class OrderCreated(Event):
    order_id: str
    customer_id: str
    items: List[dict]

@dataclass
class ItemAdded(Event):
    order_id: str
    sku: str
    quantity: int
    price: float

@dataclass
class OrderSubmitted(Event):
    order_id: str
    total: float

Aggregate und Replay

class Order:
    def __init__(self, order_id: str):
        self.order_id = order_id
        self.items = []
        self.status = "draft"
        self.total = 0
    
    def apply(self, event: Event):
        """Event auf Zustand anwenden"""
        if isinstance(event, OrderCreated):
            self.items = event.items
        elif isinstance(event, ItemAdded):
            self.items.append({
                "sku": event.sku,
                "quantity": event.quantity,
                "price": event.price
            })
        elif isinstance(event, OrderSubmitted):
            self.status = "submitted"
            self.total = event.total
    
    @classmethod
    def from_events(cls, order_id: str, events: List[Event]):
        """Zustand aus Events rekonstruieren"""
        order = cls(order_id)
        for event in events:
            order.apply(event)
        return order

Commands und Events

class OrderService:
    def __init__(self, event_store):
        self.event_store = event_store
    
    def add_item(self, order_id: str, sku: str, quantity: int, price: float):
        # 1. Aktuelle Events laden
        events = self.event_store.get_events(f"order-{order_id}")
        
        # 2. Zustand rekonstruieren
        order = Order.from_events(order_id, events)
        
        # 3. Business Logic prüfen
        if order.status != "draft":
            raise Exception("Cannot modify submitted order")
        
        # 4. Neues Event erstellen und speichern
        event = ItemAdded(
            event_id=str(uuid4()),
            timestamp=clock.now(),
            order_id=order_id,
            sku=sku,
            quantity=quantity,
            price=price
        )
        self.event_store.append(f"order-{order_id}", event)

Snapshots

Bei vielen Events kann Replay langsam werden. Eine häufige Lösung sind Snapshots.

class SnapshotStore:
    def save_snapshot(self, stream_id: str, state: dict, version: int):
        # Aktuellen Zustand speichern
        ...
    
    def get_snapshot(self, stream_id: str):
        # Letzten Snapshot laden
        ...

def load_order(order_id: str):
    # 1. Snapshot laden (falls vorhanden)
    snapshot = snapshot_store.get_snapshot(f"order-{order_id}")
    
    if snapshot:
        order = Order.from_snapshot(snapshot.state)
        # Nur Events seit Snapshot laden
        events = event_store.get_events_since(
            f"order-{order_id}", 
            snapshot.version
        )
    else:
        order = Order(order_id)
        events = event_store.get_events(f"order-{order_id}")
    
    # 2. Restliche Events anwenden
    for event in events:
        order.apply(event)
    
    return order

CQRS + Event Sourcing

Oft kombiniert:

Commands → Event Store → Events
                ↓
           Projections
                ↓
         Read Models (optimiert für Queries)

Projections: Transformieren Events in lesbare oder abfrageoptimierte Views.

class OrderListProjection:
    def __init__(self, read_db):
        self.read_db = read_db
    
    def handle(self, event: Event):
        if isinstance(event, OrderCreated):
            self.read_db.insert("orders", {
                "id": event.order_id,
                "customer": event.customer_id,
                "status": "draft"
            })
        elif isinstance(event, OrderSubmitted):
            self.read_db.update("orders", 
                {"id": event.order_id}, 
                {"status": "submitted", "total": event.total}
            )