Mock CQRS Read/Write Split in Python

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


@dataclass
class Order:
    id: int
    amount: float
    status: str = "pending"


class OrderWriteModel:
    """Handles all mutations (writes) to orders."""

    def __init__(self):
        self._orders: Dict[int, Order] = {}
        self._next_id = 1

    def create_order(self, amount: float) -> Order:
        order = Order(id=self._next_id, amount=amount)
        self._orders[order.id] = order
        self._next_id += 1
        return order

    def update_status(self, order_id: int, new_status: str) -> None:
        order = self._orders[order_id]
        order.status = new_status


class OrderReadModel:
    """Handles all queries (reads) - separate from writes."""

    def __init__(self, write_model: OrderWriteModel):
        self._orders = write_model._orders  # shared source of truth

    def get_order(self, order_id: int) -> Order:
        return self._orders[order_id]

    def list_pending(self) -> List[Order]:
        return [o for o in self._orders.values() if o.status == "pending"]


if __name__ == "__main__":
    write_model = OrderWriteModel()
    read_model = OrderReadModel(write_model)

    # Write operation
    order = write_model.create_order(amount=125.50)
    print("Created order:", order.id)

    # Read operation (separate model)
    fetched = read_model.get_order(order.id)
    print("Fetched order amount:", fetched.amount)

    # Write another update
    write_model.update_status(order.id, "completed")
    print("Pending orders after update:", [o.id for o in read_model.list_pending()])