Instead of storing current state, event sourcing stores every change as an immutable event. The current state is derived by replaying the event log — enabling time travel, audit trails, and multiple read models.
| Aspect | Event Sourcing | Traditional CRUD |
|---|---|---|
| Audit trail | Complete — every change recorded | Manual — need extra audit table |
| Time travel | Yes — replay to any point in time | No — only current state |
| Complexity | Higher — schema evolution tricky | Lower — simpler mental model |
| Performance | Writes fast; reads via projections | Reads fast with indexes |
| Best for | Financial systems, audit-heavy, complex domains | Simple CRUD, most web apps |
from dataclasses import dataclass, field
from typing import List
import json, time
@dataclass
class Event:
type: str
data: dict
timestamp: float = field(default_factory=time.time)
version: int = 0
class OrderAggregate:
def __init__(self):
self.events: List[Event] = []
self.status = "new"
self.total = 0.0
self.version = 0
def apply(self, event: Event):
"""Mutate state from event — deterministic."""
if event.type == "OrderPlaced":
self.status = "placed"
self.total = event.data["total"]
elif event.type == "PaymentReceived":
self.status = "paid"
elif event.type == "OrderShipped":
self.status = "shipped"
elif event.type == "OrderCancelled":
self.status = "cancelled"
self.version += 1
def replay(self, events: List[Event]):
"""Rebuild state from event history."""
for e in events:
self.apply(e)
def snapshot(self) -> dict:
"""Save checkpoint to avoid replaying all events."""
return {"status": self.status, "total": self.total, "version": self.version}