How to mock a CQRS projector read model update in Python

from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class OrderReadModel:
    order_id: str
    customer_name: str
    total: float
    status: str = "pending"
    items: List[Dict] = field(default_factory=list)

    def apply_event(self, event_type: str, payload: Dict) -> None:
        """Apply a domain event to update the read model (projection)."""
        if event_type == "OrderPlaced":
            self.status = "placed"
            self.items = payload.get("items", [])
            self.total = payload.get("total", 0.0)
        elif event_type == "OrderShipped":
            self.status = "shipped"
        elif event_type == "OrderCancelled":
            self.status = "cancelled"

    def __repr__(self) -> str:
        return f"OrderReadModel({self.order_id}, {self.status}, ${self.total}, {len(self.items)} items)"


class OrderProjector:
    """CQRS projector — maintains a denormalized read model from domain events."""

    def __init__(self) -> None:
        self._read_models: Dict[str, OrderReadModel] = {}

    def project(self, event: Dict) -> None:
        event_type = event["type"]
        order_id = event["order_id"]

        if event_type == "OrderPlaced":
            self._read_models[order_id] = OrderReadModel(
                order_id=order_id,
                customer_name=event["customer_name"],
                total=0.0,
            )

        read_model = self._read_models.get(order_id)
        if read_model:
            read_model.apply_event(event_type, event)
            print(f"[Projector] Applied {event_type} → {read_model}")

    def get_order(self, order_id: str) -> Optional[OrderReadModel]:
        return self._read_models.get(order_id)


if __name__ == "__main__":
    projector = OrderProjector()

    events = [
        {
            "type": "OrderPlaced",
            "order_id": "ord-001",
            "customer_name": "Alice",
            "items": [{"sku": "A1", "qty": 2}],
            "total": 49.99,
        },
        {"type": "OrderShipped", "order_id": "ord-001"},
        {"type": "OrderPlaced", "order_id": "ord-002", "customer_name": "Bob", "items": [], "total": 10.00},
    ]

    for event in events:
        projector.project(event)

    print("\nRead model snapshot:")
    for oid in ["ord-001", "ord-002", "ord-003"]:
        order = projector.get_order(oid)
        print(f"{oid}: {order if order else 'NOT FOUND'}")