CQRS

CQRS im Überblick

CQRS trennt deine Anwendung in zwei Teile: Einen für Änderungen (Commands) und einen für Abfragen (Queries). Jeder Teil kann unabhängig optimiert werden.

Traditionell (ein Modell für alles):

┌─────────────────────────────────┐
│           Application           │
│  ┌─────────────────────────┐   │
│  │      Single Model       │   │
│  │  (Read + Write)         │   │
│  └─────────────────────────┘   │
│              ↓                  │
│  ┌─────────────────────────┐   │
│  │       Database          │   │
│  └─────────────────────────┘   │
└─────────────────────────────────┘

CQRS (getrennte Modelle):

┌─────────────────────────────────────────────┐
│                Application                   │
│  ┌─────────────┐     ┌─────────────┐        │
│  │  Commands   │     │   Queries   │        │
│  │  (Write)    │     │   (Read)    │        │
│  └──────┬──────┘     └──────┬──────┘        │
│         ↓                   ↓               │
│  ┌─────────────┐     ┌─────────────┐        │
│  │ Write DB    │ ──→ │  Read DB    │        │
│  │ (normalized)│     │(denormalized)│       │
│  └─────────────┘     └─────────────┘        │
└─────────────────────────────────────────────┘

Warum trennen?

Technisch betrachtet

Commands und Queries

Command (ändert Zustand):

@dataclass
class CreateOrderCommand:
    customer_id: str
    items: List[OrderItem]

@dataclass
class AddItemCommand:
    order_id: str
    sku: str
    quantity: int

class CommandHandler:
    def handle_create_order(self, cmd: CreateOrderCommand):
        order = Order.create(cmd.customer_id, cmd.items)
        self.repository.save(order)
        # Kein Return-Wert (oder nur ID)

Query (liest Zustand):

@dataclass
class GetOrderQuery:
    order_id: str

@dataclass
class ListOrdersQuery:
    customer_id: str
    status: Optional[str] = None

class QueryHandler:
    def handle_get_order(self, query: GetOrderQuery) -> OrderDTO:
        return self.read_db.get_order(query.order_id)
    
    def handle_list_orders(self, query: ListOrdersQuery) -> List[OrderDTO]:
        return self.read_db.find_orders(
            customer_id=query.customer_id,
            status=query.status
        )

Synchronisation Write → Read

Option 1: Synchron (einfacher, aber stärker gekoppelt):

class OrderService:
    def create_order(self, cmd: CreateOrderCommand):
        # Write
        order = Order.create(cmd.customer_id, cmd.items)
        self.write_db.save(order)
        
        # Sync to Read (im selben Request)
        self.read_db.upsert_order_view(order.to_view())

Option 2: Asynchron via Events (entkoppelter, aber eventual consistent):

class OrderService:
    def create_order(self, cmd: CreateOrderCommand):
        order = Order.create(cmd.customer_id, cmd.items)
        self.write_db.save(order)
        
        # Event publizieren
        self.event_bus.publish(OrderCreatedEvent(order))

class OrderViewUpdater:
    @subscribe(OrderCreatedEvent)
    def handle(self, event: OrderCreatedEvent):
        # Read-Modell asynchron aktualisieren
        self.read_db.upsert_order_view(event.order.to_view())

Read-Modell optimieren

Write-Modell (normalisiert):

-- Mehrere Tabellen, Joins nötig
orders (id, customer_id, status, created_at)
order_items (id, order_id, product_id, quantity, price)
products (id, name, description)
customers (id, name, email)

Read-Modell (denormalisiert):

-- Eine Tabelle, alles drin
order_views (
    id,
    customer_name,
    customer_email,
    status,
    items_json,  -- [{name, quantity, price}, ...]
    total,
    created_at
)

Eventual Consistency

Timeline:
─────────────────────────────────────────────────→
    │                    │                    │
    ▼                    ▼                    ▼
  Write              Event              Read Updated
                    Published
    
    └────── mögliche Inconsistency Window ──────┘

Umgang damit:

• UI zeigt optimistisch den erwarteten Zustand
• Polling oder WebSocket für Updates
• Für kritische Reads: Direkt aus Write-DB lesen

Wann CQRS?