CQRS + Event Sourcing Architecture Patterns

As systems scale, traditional CRUD single-model architectures hit bottlenecks in complex queries, audit trails, and concurrency. CQRS + ES are advanced patterns that break these limits, but added complexity demands careful adoption decisions.

Event Sourcing: Events as Source of Truth

Never persist current state; append only events that caused state changes (OrderCreated → ItemAdded → OrderPlaced). Replaying reconstructs state at any point; complete audit trail for free. Pair with snapshots to avoid replaying from zero on each load.

from __future__ import annotations
import json, time
from dataclasses import dataclass, field
from typing import Callable, Iterable, Any
from collections import defaultdict

@dataclass
class DomainEvent:
    event_id: str
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: dict
    occurred_at: float = field(default_factory=time.time)
    version: int = 1

class EventStore:
    def __init__(self) -> None:
        self._events: list[DomainEvent] = []
        self._snapshots: dict[tuple[str, str], tuple[int, Any]] = {}
        self._subs: dict[str, list[Callable[[DomainEvent], None]]] = defaultdict(list)

    def append(self, events: Iterable[DomainEvent], expected_version: int | None = None) -> None:
        events = list(events)
        if expected_version is not None and events:
            at, aid = events[0].aggregate_type, events[0].aggregate_id
            cur = sum(1 for e in self._events if e.aggregate_type == at and e.aggregate_id == aid)
            if cur != expected_version:
                raise ValueError(f"Concurrency conflict: expected v{expected_version}, got v{cur}")
        for e in events:
            self._events.append(e)
            for s in self._subs.get(e.event_type, []) + self._subs.get("*", []): s(e)

    def load(self, agg_type: str, agg_id: str, snapshot_every: int = 50) -> tuple[Any, int]:
        key = (agg_type, agg_id)
        start_ver, state = self._snapshots.get(key, (0, None))
        version = start_ver
        events = [e for e in self._events if e.aggregate_type == agg_type and e.aggregate_id == agg_id][start_ver:]
        for e in events:
            version += 1
            fn = globals().get(f"apply_{e.event_type}")
            state = fn(state, e) if fn else state
            if version % snapshot_every == 0: self._snapshots[key] = (version, state)
        return state, version

    def subscribe(self, event_type: str, fn: Callable[[DomainEvent], None]) -> None:
        self._subs[event_type].append(fn)

def apply_OrderCreated(state, e): return {"order_id": e.aggregate_id, "items": {}, "status": "CREATED", **e.payload}
def apply_ItemAdded(state, e):
    s = dict(state); pid = e.payload["product_id"]
    s["items"] = {**s["items"], pid: s["items"].get(pid, 0) + e.payload["qty"]}
    return s
def apply_OrderPlaced(state, e): return {**state, "status": "PLACED", "placed_at": e.occurred_at}

store = EventStore()
projection: dict[str, dict] = {}
store.subscribe("OrderPlaced", lambda e: projection.__setitem__(e.aggregate_id, {"status": "PLACED", **e.payload}))

CQRS: Separate Write and Read Models

Write path goes through domain model and produces events. Event subscribers asynchronously build denormalized projection tables tuned for queries. Write side enforces consistency while read side optimized for retrieval, but introduces an eventual consistency window.

Best Practices

Ask three questions before adopting: Do you require an immutable audit trail? Are queries genuinely painful with CRUD? Does the team have DDD experience? Only proceed if all three are Yes. Otherwise classic CRUD plus a simple audit table nearly always delivers better ROI.