How to Implement CQRS with Separate Read and Write Models in Python

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


@dataclass
class OrderWriteModel:
    order_id: int
    customer: str
    items: List[str] = field(default_factory=list)

    def add_item(self, item: str) -> None:
        self.items.append(item)


@dataclass
class OrderReadModel:
    order_id: int
    customer: str
    item_count: int
    total_price: float


class OrderWriteRepository:
    def __init__(self) -> None:
        self._orders: Dict[int, OrderWriteModel] = {}

    def save(self, order: OrderWriteModel) -> None:
        self._orders[order.order_id] = order

    def get(self, order_id: int) -> Optional[OrderWriteModel]:
        return self._orders.get(order_id)


class OrderReadRepository:
    def __init__(self) -> None:
        self._orders: Dict[int, OrderReadModel] = {}

    def save(self, order: OrderReadModel) -> None:
        self._orders[order.order_id] = order

    def get_summary(self, order_id: int) -> Optional[OrderReadModel]:
        return self._orders.get(order_id)

    def all_summaries(self) -> List[OrderReadModel]:
        return list(self._orders.values())


class OrderService:
    PRICE_PER_ITEM = 10.0

    def __init__(self, write_repo: OrderWriteRepository, read_repo: OrderReadRepository) -> None:
        self.write_repo = write_repo
        self.read_repo = read_repo

    def place_order(self, order_id: int, customer: str, items: List[str]) -> None:
        order = OrderWriteModel(order_id=order_id, customer=customer, items=items)
        self.write_repo.save(order)

        read_model = OrderReadModel(
            order_id=order_id,
            customer=customer,
            item_count=len(items),
            total_price=len(items) * self.PRICE_PER_ITEM,
        )
        self.read_repo.save(read_model)


def main() -> None:
    write_repo = OrderWriteRepository()
    read_repo = OrderReadRepository()
    service = OrderService(write_repo, read_repo)

    service.place_order(1, "Alice", ["Laptop", "Mouse", "Keyboard"])
    service.place_order(2, "Bob", ["Monitor"])

    for order in read_repo.all_summaries():
        print(f"Order {order.order_id} by {order.customer}: "
              f"{order.item_count} items, total ${order.total_price:.2f}")

    write_order = write_repo.get(1)
    print(f"Write model items for order 1: {write_order.items}")


if __name__ == "__main__":
    main()