CQRS with Separate Read and Write Repositories in Python

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


# --- Write side: commands mutate state ---
@dataclass
class CreateUserCommand:
    id: int
    name: str


class UserWriteRepository:
    def __init__(self) -> None:
        self._store: Dict[int, Dict[str, object]] = {}

    def create(self, command: CreateUserCommand) -> None:
        self._store[command.id] = {"id": command.id, "name": command.name}

    def update_name(self, user_id: int, new_name: str) -> None:
        if user_id in self._store:
            self._store[user_id]["name"] = new_name

    def delete(self, user_id: int) -> None:
        self._store.pop(user_id, None)


# --- Read side: queries return DTOs, never mutate ---
@dataclass(frozen=True)
class UserReadModel:
    id: int
    name: str


class UserReadRepository:
    def __init__(self, write_repo: UserWriteRepository) -> None:
        self._write_repo = write_repo

    def get_by_id(self, user_id: int) -> Optional[UserReadModel]:
        data = self._write_repo._store.get(user_id)
        if data is None:
            return None
        return UserReadModel(id=data["id"], name=data["name"])

    def list_all(self) -> List[UserReadModel]:
        return [
            UserReadModel(id=data["id"], name=data["name"])
            for data in self._write_repo._store.values()
        ]


if __name__ == "__main__":
    write_repo = UserWriteRepository()
    read_repo = UserReadRepository(write_repo)

    # Write operations
    write_repo.create(CreateUserCommand(id=1, name="Alice"))
    write_repo.create(CreateUserCommand(id=2, name="Bob"))
    write_repo.update_name(1, "Alicia")

    # Read operations — pure queries
    user = read_repo.get_by_id(1)
    all_users = read_repo.list_all()

    print(f"User 1: {user}")
    print(f"All: {all_users}")