Event sourcing fundamentals

Need to maintain a complete audit trail, support temporal queries, and enable rebuilding state from history.

Event sourcing stores state changes as immutable events:

# Event store
@dataclass
class Event:
    aggregate_id: str
    event_type: str
    data: dict
    timestamp: datetime
    version: int

# Events are facts - things that happened
events = [
    Event('order-123', 'OrderCreated', {'customer': 'alice', 'items': [...]}, ...),
    Event('order-123', 'ItemAdded', {'sku': 'WIDGET-1', 'qty': 2}, ...),
    Event('order-123', 'OrderPaid', {'amount': 49.99, 'method': 'card'}, ...),
    Event('order-123', 'OrderShipped', {'tracking': 'TRK-456'}, ...),
]

# Rebuild current state by replaying events
def rebuild_order(events):
    order = {}
    for event in events:
        if event.event_type == 'OrderCreated':
            order = {'id': event.aggregate_id, 'status': 'created', **event.data}
        elif event.event_type == 'ItemAdded':
            order.setdefault('items', []).append(event.data)
        elif event.event_type == 'OrderPaid':
            order['status'] = 'paid'
        elif event.event_type == 'OrderShipped':
            order['status'] = 'shipped'
    return order

When to use:

• Audit requirements (finance, healthcare, compliance)
• Need to answer 'what was the state at time T?'
• Complex domain with many state transitions
• Need to rebuild read models from different perspectives

When NOT to use:

• Simple CRUD applications
• No audit requirements
• Team is not familiar with the pattern
• Low write volume doesn't justify complexity

Traditional databases store current state and discard history. Event sourcing preserves the complete history, enabling audit trails, debugging, and replaying events to build new views.

Systems requiring complete audit trails or complex state management