CQRS Pattern: A Complete Guide with Examples in 11 Programming Languages

CQRS — Command Query Responsibility Segregation — is an architectural pattern that separates the operations that change state from the operations that read state into two distinct models. Every operation is either a Command (it changes data and returns no result) or a Query (it reads data and returns a result but changes nothing). The two models are handled by separate objects, separate code paths, and optionally separate data stores.

The principle behind CQRS was articulated by Bertrand Meyer as the Command-Query Separation (CQS) principle: a method should either do something (command) or answer something (query), but never both. CQRS scales this principle from individual methods to the architecture of an entire application — giving the read side and the write side their own models, their own optimizations, and their own contracts.

In this comprehensive guide, we'll explore the CQRS pattern with real-world examples in Python, TypeScript, Java, JavaScript, C#, PHP, Go, Rust, Dart, Swift, and Kotlin — complete with language-specific best practices, common pitfalls, and when to use this powerful pattern.

Table of Contents

• What is CQRS?
• Why Use CQRS?
• CQRS Comparison
• CQRS Explained
• Architecture Diagram
• Beginner-Friendly Example
• Production-Ready Example
• Real-World Use Cases
• Language-Specific Mistakes and Anti-Patterns
• Frequently Asked Questions
• Key Takeaways

What is CQRS?

Most applications start with a single model that handles everything. A UserService has createUser(), updateUser(), deleteUser(), and getUser() — both writes and reads go through the same service, the same repository, and the same database model. This works well for simple CRUD applications, but it begins to create friction as applications grow.

The problem is that reads and writes have fundamentally different needs. Writes are concerned with domain logic: invariants must be enforced, concurrency must be handled, state transitions must be valid. The write model benefits from rich domain objects — aggregates, value objects, domain events — that encapsulate business rules. Reads are concerned with presentation: they return data shaped exactly how the UI needs it, often joining many tables, applying filters, and projecting specific fields. The read model benefits from denormalized, flat structures optimized for querying — not for enforcing business rules.

When reads and writes share the same model, each compromises the other. The domain model becomes polluted with query convenience methods. The query layer must navigate complex domain relationships to produce simple view data. Scaling reads means scaling the write infrastructure, and vice versa. Adding a new query view requires modifying the shared model.

CQRS resolves this by splitting the model in two. The Write Side owns the domain model — commands flow in, domain logic executes, state changes. The Read Side owns the query model — queries flow in, data returns, nothing changes. The two sides communicate: when the write side changes state, the read side is notified and updates its own projections. This communication can be synchronous (in-process, in simple CQRS) or asynchronous (via events, in full CQRS with Event Sourcing).

Why Use CQRS?

CQRS earns its complexity when any of these conditions are true:

1. Different scaling requirements: Read traffic vastly outnumbers write traffic (as it does in most applications). CQRS lets you scale the read side independently — replicas, caches, read-optimized stores — without scaling the write infrastructure.

2. Different optimization needs: Write models are normalized for data integrity; read models are denormalized for query performance. A single shared model forces a compromise that serves neither well.

3. Complex domain logic: When the write side has significant business rules, keeping them separate from query concerns makes each simpler and easier to reason about. The domain model is not polluted with query convenience methods.

4. Multiple read representations: The same data may need to be presented in many different shapes — a product detail page, a product list, a search result, an admin dashboard. Each read model can be independently optimized for its consumer without affecting the write model.

5. Audit and event history: CQRS pairs naturally with Event Sourcing — the write side records every state change as an event. The full history of every change is preserved, enabling audit trails, time-travel debugging, and the ability to rebuild any read model from scratch.

6. Team scalability: Separate read and write sides can be developed by separate teams or deployed to separate services. The write team works on domain logic; the read team works on query optimization.

CQRS Comparison

CQRS vs. Repository Pattern: The Repository Pattern separates data access from domain logic but still uses a single model for both reads and writes. CQRS goes further — it separates the model itself, giving reads and writes completely different objects, data shapes, and optimization paths.

CQRS vs. Active Record: Active Record collapses everything into a single model object that is both the domain model and the query model. CQRS is the explicit rejection of this: the domain model (write side) and the query model (read side) are different things, with different shapes and different purposes.

CQRS vs. Event-Driven Architecture: EDA and CQRS are complementary and often combined. CQRS defines the read/write split; EDA defines how the write side notifies the read side of changes. Full CQRS typically uses domain events to propagate changes from the write side to read-side projections — making it an EDA implementation at the application level.

Simple CQRS vs. CQRS + Event Sourcing: Simple CQRS separates handlers but may still use the same data store. CQRS + Event Sourcing replaces the write-side state with an append-only log of events — the current state is derived by replaying the event log. Event Sourcing adds full history and audit capability but also adds significant complexity.

CQRS Explained

Core Components

Command: An object that represents an intention to change state. Commands are named in the imperative: PlaceOrderCommand, UpdateUserEmailCommand, CancelSubscriptionCommand. A command carries the data needed to perform the change. It does not return a result — it may succeed or fail, but it does not return data.

Command Handler: Receives a command, validates it, loads the relevant domain aggregate, executes the business logic, persists the changed state, and optionally publishes domain events. One handler per command. The handler is the boundary of a single write transaction.

Query: An object that represents a request for data. Queries are named as questions: GetOrderByIdQuery, ListActiveUsersQuery, SearchProductsQuery. A query carries the parameters needed to filter and shape the data. It returns data and changes nothing.

Query Handler: Receives a query and returns data. The query handler bypasses the domain model entirely — it queries the data store directly, projecting exactly the fields the caller needs, joining whatever tables are necessary, applying filters and pagination. One handler per query.

Command Bus / Query Bus: An optional dispatcher that routes commands and queries to their respective handlers. The bus decouples the caller from the handler — the caller sends to the bus, the bus finds the right handler. This enables middleware (logging, validation, authorization, transactions) to be applied uniformly to all commands or queries.

Read Model / Projection: A data structure optimized for reading. It may be a denormalized database table, a document in a NoSQL store, a cache entry, or an in-memory structure. It is updated whenever the write side changes state — synchronously in simple CQRS, or asynchronously via domain events in full CQRS.

Command-Query Separation at the Method Level (CQS)

Before applying CQRS at the architectural level, apply it at the method level. Every method should either:

• Command: Change state. Return void (or a task/promise equivalent). Name it as a verb: save(), delete(), process(), publish().
• Query: Return data. Change nothing. Name it as a question or description: get(), find(), list(), count(), isValid().

Never mix the two: saveAndReturn() that saves and returns the saved object, or pop() that removes and returns — both violate CQS. In practice, some pragmatic exceptions exist (creating an entity and returning its ID is a common one), but the discipline of thinking "is this a command or a query?" before writing any method is immediately valuable.

CQRS Variants

Simple CQRS (Logical separation): The read and write models are separate objects in the same application, using the same database. Commands go to domain objects via command handlers; queries go directly to the database via query handlers, using SQL or ORM projections. This is the most practical starting point — significant benefit, manageable complexity.

CQRS with Separate Read Store: The write side uses a normalized relational database; the read side uses a separate, denormalized store (a read-optimized SQL schema, a document store, a search index, a cache). Domain events propagate changes from the write store to the read store asynchronously. This enables maximum read optimization but adds eventual consistency.

CQRS + Event Sourcing: The write side stores events — not current state. Every state change is appended as an immutable event to an event log. The current state of any aggregate is derived by replaying its events. The read side subscribes to those events and builds projections. This provides a complete audit trail and the ability to rebuild any projection at any time, but requires significant infrastructure and a mindset shift.

Eventual Consistency

Full CQRS with async event propagation introduces eventual consistency: after a command succeeds, the read model may not yet reflect the change. A user places an order; the command succeeds; but the "order list" query may still return the old data for a brief moment while the projection updates.

This is acceptable in many real-world scenarios — user sees "Order placed!" and then sees their order in the list a moment later. It is not acceptable in all scenarios — immediately after placing an order, the user must not see a count of 0 orders when the count is clearly 1.

Strategies for managing eventual consistency: return the updated data from the command handler for immediate display; use correlation IDs to poll for projection updates; use synchronous projection updates for critical reads while keeping async updates for non-critical views.

Architecture Diagram

Beginner-Friendly Example

The clearest CQRS introduction: a simple task management system where tasks can be created, completed, and deleted (commands) and listed or fetched by ID (queries). The command side handles domain logic; the query side returns flat data structures optimized for display.

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional
from uuid import uuid4


# ════════════════════════════════════════════════════════════════
# Commands — intentions to change state; carry no return value
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class CreateTaskCommand:
    title:       str
    description: str
    priority:    str = "medium"   # "low" | "medium" | "high"

@dataclass(frozen=True)
class CompleteTaskCommand:
    task_id: str

@dataclass(frozen=True)
class DeleteTaskCommand:
    task_id: str


# ════════════════════════════════════════════════════════════════
# Queries — requests for data; change nothing
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class GetTaskByIdQuery:
    task_id: str

@dataclass(frozen=True)
class ListTasksQuery:
    status:   Optional[str] = None   # None = all | "pending" | "completed"
    priority: Optional[str] = None


# ════════════════════════════════════════════════════════════════
# Write Side — domain model (rich, enforces business rules)
# ════════════════════════════════════════════════════════════════

@dataclass
class Task:
    """Domain aggregate — owns business logic and invariants."""
    id:          str
    title:       str
    description: str
    priority:    str
    status:      str      # "pending" | "completed" | "deleted"
    created_at:  datetime
    completed_at: Optional[datetime] = None

    def complete(self) -> None:
        if self.status != "pending":
            raise ValueError(f"Cannot complete a task with status '{self.status}'")
        self.status       = "completed"
        self.completed_at = datetime.now(tz=timezone.utc)

    def delete(self) -> None:
        if self.status == "deleted":
            raise ValueError("Task is already deleted")
        self.status = "deleted"


# In-memory "database" — separate write store and read store
_write_store: dict[str, Task] = {}   # write side: domain objects


# ════════════════════════════════════════════════════════════════
# Read Side — view models (flat, shaped for display)
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class TaskSummaryView:
    """Flat read model — shaped for list display."""
    id:       str
    title:    str
    priority: str
    status:   str

@dataclass(frozen=True)
class TaskDetailView:
    """Flat read model — shaped for detail display."""
    id:           str
    title:        str
    description:  str
    priority:     str
    status:       str
    created_at:   str
    completed_at: Optional[str]


# In-memory read store — denormalized, shaped for queries
_read_store: dict[str, dict] = {}   # read side: flat projections


def _update_projection(task: Task) -> None:
    """Synchronizes the read store after a write-side change."""
    if task.status == "deleted":
        _read_store.pop(task.id, None)
        return
    _read_store[task.id] = {
        "id":           task.id,
        "title":        task.title,
        "description":  task.description,
        "priority":     task.priority,
        "status":       task.status,
        "created_at":   task.created_at.isoformat(),
        "completed_at": task.completed_at.isoformat() if task.completed_at else None,
    }


# ════════════════════════════════════════════════════════════════
# Command Handlers — write side logic
# ════════════════════════════════════════════════════════════════

class CreateTaskHandler:
    def handle(self, cmd: CreateTaskCommand) -> str:
        """Returns the new task ID — pragmatic exception to 'commands return nothing'."""
        if not cmd.title.strip():
            raise ValueError("Task title cannot be empty")
        if cmd.priority not in ("low", "medium", "high"):
            raise ValueError(f"Invalid priority: {cmd.priority}")

        task = Task(
            id=str(uuid4())[:8],
            title=cmd.title.strip(),
            description=cmd.description.strip(),
            priority=cmd.priority,
            status="pending",
            created_at=datetime.now(tz=timezone.utc),
        )
        _write_store[task.id] = task
        _update_projection(task)   # sync read store
        print(f"  [Write] Task created: {task.id} — '{task.title}'")
        return task.id


class CompleteTaskHandler:
    def handle(self, cmd: CompleteTaskCommand) -> None:
        task = _write_store.get(cmd.task_id)
        if not task:
            raise KeyError(f"Task not found: {cmd.task_id}")
        task.complete()              # domain logic enforced here
        _update_projection(task)     # sync read store
        print(f"  [Write] Task completed: {cmd.task_id}")


class DeleteTaskHandler:
    def handle(self, cmd: DeleteTaskCommand) -> None:
        task = _write_store.get(cmd.task_id)
        if not task:
            raise KeyError(f"Task not found: {cmd.task_id}")
        task.delete()
        _update_projection(task)     # removes from read store
        print(f"  [Write] Task deleted: {cmd.task_id}")


# ════════════════════════════════════════════════════════════════
# Query Handlers — read side, bypass domain model entirely
# ════════════════════════════════════════════════════════════════

class GetTaskByIdHandler:
    def handle(self, query: GetTaskByIdQuery) -> Optional[TaskDetailView]:
        data = _read_store.get(query.task_id)
        if not data:
            return None
        return TaskDetailView(**data)


class ListTasksHandler:
    def handle(self, query: ListTasksQuery) -> list[TaskSummaryView]:
        results = list(_read_store.values())

        if query.status:
            results = [r for r in results if r["status"] == query.status]
        if query.priority:
            results = [r for r in results if r["priority"] == query.priority]

        # Sort: high priority first, then by creation time
        priority_order = {"high": 0, "medium": 1, "low": 2}
        results.sort(key=lambda r: (priority_order.get(r["priority"], 1), r["created_at"]))

        return [TaskSummaryView(id=r["id"], title=r["title"],
                                priority=r["priority"], status=r["status"])
                for r in results]


# ════════════════════════════════════════════════════════════════
# Command Bus & Query Bus — optional dispatchers
# ════════════════════════════════════════════════════════════════

class CommandBus:
    def __init__(self) -> None:
        self._handlers: dict[type, object] = {}

    def register(self, command_type: type, handler: object) -> None:
        self._handlers[command_type] = handler

    def send(self, command: object):
        handler = self._handlers.get(type(command))
        if not handler:
            raise RuntimeError(f"No handler for {type(command).__name__}")
        return handler.handle(command)


class QueryBus:
    def __init__(self) -> None:
        self._handlers: dict[type, object] = {}

    def register(self, query_type: type, handler: object) -> None:
        self._handlers[query_type] = handler

    def ask(self, query: object):
        handler = self._handlers.get(type(query))
        if not handler:
            raise RuntimeError(f"No handler for {type(query).__name__}")
        return handler.handle(query)


# ════════════════════════════════════════════════════════════════
# Wiring
# ════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    # Wire command bus
    cmd_bus = CommandBus()
    cmd_bus.register(CreateTaskCommand,   CreateTaskHandler())
    cmd_bus.register(CompleteTaskCommand, CompleteTaskHandler())
    cmd_bus.register(DeleteTaskCommand,   DeleteTaskHandler())

    # Wire query bus
    qry_bus = QueryBus()
    qry_bus.register(GetTaskByIdQuery, GetTaskByIdHandler())
    qry_bus.register(ListTasksQuery,   ListTasksHandler())

    print("=== Commands (write side) ===")
    task_a = cmd_bus.send(CreateTaskCommand("Deploy to production", "Run deployment pipeline", "high"))
    task_b = cmd_bus.send(CreateTaskCommand("Write unit tests",     "Cover new features",      "medium"))
    task_c = cmd_bus.send(CreateTaskCommand("Update README",        "Document new API",        "low"))

    cmd_bus.send(CompleteTaskCommand(task_a))

    print("\n=== Queries (read side) ===")

    # Detail view
    detail = qry_bus.ask(GetTaskByIdQuery(task_a))
    print(f"\nDetail for {task_a}:")
    print(f"  Title:    {detail.title}")
    print(f"  Status:   {detail.status}")
    print(f"  Completed:{detail.completed_at[:19] if detail.completed_at else 'N/A'}")

    # List all
    all_tasks = qry_bus.ask(ListTasksQuery())
    print(f"\nAll tasks ({len(all_tasks)}):")
    for t in all_tasks:
        print(f"  [{t.priority.upper():6}] {t.title:<35} ({t.status})")

    # List pending only
    pending = qry_bus.ask(ListTasksQuery(status="pending"))
    print(f"\nPending tasks ({len(pending)}):")
    for t in pending:
        print(f"  {t.title}")

    # Delete a task — disappears from read store
    print(f"\n=== Deleting task {task_c} ===")
    cmd_bus.send(DeleteTaskCommand(task_c))
    all_after = qry_bus.ask(ListTasksQuery())
    print(f"Tasks after deletion: {len(all_after)}")

    # Try to complete an already-completed task — domain error
    print("\n=== Domain rule enforcement ===")
    try:
        cmd_bus.send(CompleteTaskCommand(task_a))
    except ValueError as e:
        print(f"  Caught expected error: {e}")

Production-Ready Example

The most instructive CQRS challenge in production: an e-commerce order system with a write side that enforces complex domain invariants (inventory reservation, payment validation, status transitions) and a read side with multiple independent projections (order detail view, customer order history, fulfillment dashboard), updated via domain events.

🐍 Python (Production)

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Optional, Callable
from uuid import uuid4


# ════════════════════════════════════════════════════════════════
# Domain events — published by the write side; consumed by projections
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class DomainEvent:
    event_id:   str = field(default_factory=lambda: str(uuid4())[:8])
    occurred_at: datetime = field(default_factory=lambda: datetime.now(tz=timezone.utc))

@dataclass(frozen=True)
class OrderPlacedEvent(DomainEvent):
    order_id:    str = ""
    customer_id: str = ""
    items:       tuple = ()
    total:       str   = "0.00"    # Decimal serialized as string for immutability

@dataclass(frozen=True)
class OrderPaidEvent(DomainEvent):
    order_id:       str = ""
    payment_id:     str = ""
    amount:         str = "0.00"

@dataclass(frozen=True)
class OrderShippedEvent(DomainEvent):
    order_id:        str = ""
    tracking_number: str = ""

@dataclass(frozen=True)
class OrderCancelledEvent(DomainEvent):
    order_id: str = ""
    reason:   str = ""


# ════════════════════════════════════════════════════════════════
# Commands
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class PlaceOrderCommand:
    customer_id: str
    items:       list[dict]   # [{"sku": "SKU-001", "qty": 2, "price": "9.99"}]

@dataclass(frozen=True)
class PayOrderCommand:
    order_id:   str
    payment_id: str
    amount:     Decimal

@dataclass(frozen=True)
class ShipOrderCommand:
    order_id:        str
    tracking_number: str

@dataclass(frozen=True)
class CancelOrderCommand:
    order_id: str
    reason:   str


# ════════════════════════════════════════════════════════════════
# Queries
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class GetOrderDetailQuery:
    order_id: str

@dataclass(frozen=True)
class GetCustomerOrdersQuery:
    customer_id: str
    status:      Optional[str] = None

@dataclass(frozen=True)
class GetFulfillmentQueueQuery:
    status: str = "paid"   # orders ready to ship


# ════════════════════════════════════════════════════════════════
# Write Side — domain model with rich business logic
# ════════════════════════════════════════════════════════════════

class OrderStatus(str, Enum):
    PENDING   = "pending"
    PAID      = "paid"
    SHIPPED   = "shipped"
    CANCELLED = "cancelled"


@dataclass
class OrderItem:
    sku:   str
    qty:   int
    price: Decimal

    @property
    def subtotal(self) -> Decimal:
        return self.price * self.qty


@dataclass
class Order:
    """
    Domain Aggregate — owns all order business rules and invariants.
    All state transitions are guarded. No public setters.
    """
    id:          str
    customer_id: str
    items:       list[OrderItem]
    status:      OrderStatus = OrderStatus.PENDING
    payment_id:  Optional[str] = None
    tracking_number: Optional[str] = None
    created_at:  datetime = field(default_factory=lambda: datetime.now(tz=timezone.utc))
    _events:     list[DomainEvent] = field(default_factory=list, init=False, repr=False)

    @classmethod
    def place(cls, customer_id: str, raw_items: list[dict]) -> "Order":
        if not raw_items:
            raise ValueError("Order must contain at least one item")
        items = [OrderItem(sku=i["sku"], qty=int(i["qty"]),
                           price=Decimal(str(i["price"]))) for i in raw_items]
        if any(item.qty <= 0 for item in items):
            raise ValueError("Item quantity must be positive")

        order = cls(
            id=str(uuid4())[:8],
            customer_id=customer_id,
            items=items,
        )
        order._events.append(OrderPlacedEvent(
            order_id=order.id, customer_id=customer_id,
            items=tuple({"sku": i.sku, "qty": i.qty, "price": str(i.price)} for i in items),
            total=str(order.total),
        ))
        return order

    @property
    def total(self) -> Decimal:
        return sum(item.subtotal for item in self.items)

    def pay(self, payment_id: str, amount: Decimal) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot pay order in status '{self.status}'")
        if amount < self.total:
            raise ValueError(f"Payment {amount} insufficient for total {self.total}")
        self.status     = OrderStatus.PAID
        self.payment_id = payment_id
        self._events.append(OrderPaidEvent(
            order_id=self.id, payment_id=payment_id, amount=str(amount)
        ))

    def ship(self, tracking_number: str) -> None:
        if self.status != OrderStatus.PAID:
            raise ValueError(f"Cannot ship order in status '{self.status}'")
        self.status          = OrderStatus.SHIPPED
        self.tracking_number = tracking_number
        self._events.append(OrderShippedEvent(
            order_id=self.id, tracking_number=tracking_number
        ))

    def cancel(self, reason: str) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("Cannot cancel a shipped order")
        if self.status == OrderStatus.CANCELLED:
            raise ValueError("Order is already cancelled")
        self.status = OrderStatus.CANCELLED
        self._events.append(OrderCancelledEvent(order_id=self.id, reason=reason))

    def collect_events(self) -> list[DomainEvent]:
        """Drain and return unpublished domain events."""
        events = list(self._events)
        self._events.clear()
        return events


# ════════════════════════════════════════════════════════════════
# Read Side — three independent projections, each optimized for its consumer
# ════════════════════════════════════════════════════════════════

@dataclass
class OrderDetailView:
    """Shaped for the order detail page — everything the UI needs."""
    order_id:        str
    customer_id:     str
    status:          str
    total:           str
    items:           list[dict]
    payment_id:      Optional[str]
    tracking_number: Optional[str]
    created_at:      str


@dataclass
class CustomerOrderSummary:
    """Shaped for the customer's order history list — minimal data."""
    order_id:   str
    status:     str
    total:      str
    item_count: int
    created_at: str


@dataclass
class FulfillmentView:
    """Shaped for the warehouse fulfillment queue — what ops needs."""
    order_id:   str
    customer_id: str
    item_lines: list[str]    # ["2x SKU-001", "1x SKU-003"]
    total:      str
    paid_at:    str


# ════════════════════════════════════════════════════════════════
# Projection stores
# ════════════════════════════════════════════════════════════════

_order_detail_store:    dict[str, OrderDetailView]          = {}
_customer_orders_store: dict[str, list[CustomerOrderSummary]] = {}  # customer_id → [orders]
_fulfillment_store:     dict[str, FulfillmentView]          = {}


# ════════════════════════════════════════════════════════════════
# Event handlers — update read-side projections from domain events
# ════════════════════════════════════════════════════════════════

def _on_order_placed(event: OrderPlacedEvent) -> None:
    """Build initial projections when an order is placed."""
    total = sum(Decimal(i["price"]) * i["qty"] for i in event.items)

    _order_detail_store[event.order_id] = OrderDetailView(
        order_id=event.order_id, customer_id=event.customer_id,
        status="pending", total=str(total),
        items=list(event.items), payment_id=None, tracking_number=None,
        created_at=event.occurred_at.isoformat(),
    )

    summary = CustomerOrderSummary(
        order_id=event.order_id, status="pending", total=str(total),
        item_count=len(event.items), created_at=event.occurred_at.isoformat(),
    )
    _customer_orders_store.setdefault(event.customer_id, []).append(summary)
    print(f"  [Projection] OrderDetail + CustomerHistory updated for {event.order_id}")


def _on_order_paid(event: OrderPaidEvent) -> None:
    """Update projections when payment is confirmed; add to fulfillment queue."""
    if view := _order_detail_store.get(event.order_id):
        view.status     = "paid"
        view.payment_id = event.payment_id

    if summaries := _customer_orders_store.get(
        _order_detail_store[event.order_id].customer_id if event.order_id in _order_detail_store else ""
    ):
        for s in summaries:
            if s.order_id == event.order_id:
                object.__setattr__(s, "status", "paid") if isinstance(s, tuple) else setattr(s, "status", "paid")

    if view := _order_detail_store.get(event.order_id):
        item_lines = [f"{i['qty']}x {i['sku']}" for i in view.items]
        _fulfillment_store[event.order_id] = FulfillmentView(
            order_id=event.order_id, customer_id=view.customer_id,
            item_lines=item_lines, total=view.total,
            paid_at=event.occurred_at.isoformat(),
        )
    print(f"  [Projection] FulfillmentQueue + OrderDetail updated for {event.order_id}")


def _on_order_shipped(event: OrderShippedEvent) -> None:
    if view := _order_detail_store.get(event.order_id):
        view.status          = "shipped"
        view.tracking_number = event.tracking_number
    _fulfillment_store.pop(event.order_id, None)   # remove from fulfillment queue
    print(f"  [Projection] Order {event.order_id} shipped — removed from fulfillment queue")


def _on_order_cancelled(event: OrderCancelledEvent) -> None:
    if view := _order_detail_store.get(event.order_id):
        view.status = "cancelled"
    _fulfillment_store.pop(event.order_id, None)
    print(f"  [Projection] Order {event.order_id} cancelled")


# Simple in-process event bus
_event_handlers: dict[type, list[Callable]] = {
    OrderPlacedEvent:    [_on_order_placed],
    OrderPaidEvent:      [_on_order_paid],
    OrderShippedEvent:   [_on_order_shipped],
    OrderCancelledEvent: [_on_order_cancelled],
}

def _publish_events(events: list[DomainEvent]) -> None:
    for event in events:
        for handler in _event_handlers.get(type(event), []):
            handler(event)


# ════════════════════════════════════════════════════════════════
# Write-side command handlers
# ════════════════════════════════════════════════════════════════

_order_store: dict[str, Order] = {}   # write store: domain aggregates

class PlaceOrderHandler:
    def handle(self, cmd: PlaceOrderCommand) -> str:
        order = Order.place(cmd.customer_id, cmd.items)
        _order_store[order.id] = order
        _publish_events(order.collect_events())
        print(f"  [Write] Order placed: {order.id} (total: {order.total})")
        return order.id


class PayOrderHandler:
    def handle(self, cmd: PayOrderCommand) -> None:
        order = _order_store.get(cmd.order_id)
        if not order: raise KeyError(f"Order not found: {cmd.order_id}")
        order.pay(cmd.payment_id, cmd.amount)
        _publish_events(order.collect_events())
        print(f"  [Write] Order paid: {cmd.order_id}")


class ShipOrderHandler:
    def handle(self, cmd: ShipOrderCommand) -> None:
        order = _order_store.get(cmd.order_id)
        if not order: raise KeyError(f"Order not found: {cmd.order_id}")
        order.ship(cmd.tracking_number)
        _publish_events(order.collect_events())
        print(f"  [Write] Order shipped: {cmd.order_id}")


class CancelOrderHandler:
    def handle(self, cmd: CancelOrderCommand) -> None:
        order = _order_store.get(cmd.order_id)
        if not order: raise KeyError(f"Order not found: {cmd.order_id}")
        order.cancel(cmd.reason)
        _publish_events(order.collect_events())
        print(f"  [Write] Order cancelled: {cmd.order_id}")


# ════════════════════════════════════════════════════════════════
# Query handlers — read side only; no write-side access
# ════════════════════════════════════════════════════════════════

class GetOrderDetailHandler:
    def handle(self, query: GetOrderDetailQuery) -> Optional[OrderDetailView]:
        return _order_detail_store.get(query.order_id)


class GetCustomerOrdersHandler:
    def handle(self, query: GetCustomerOrdersQuery) -> list[CustomerOrderSummary]:
        orders = _customer_orders_store.get(query.customer_id, [])
        if query.status:
            orders = [o for o in orders if o.status == query.status]
        return sorted(orders, key=lambda o: o.created_at, reverse=True)


class GetFulfillmentQueueHandler:
    def handle(self, query: GetFulfillmentQueueQuery) -> list[FulfillmentView]:
        return sorted(_fulfillment_store.values(), key=lambda v: v.paid_at)


# ════════════════════════════════════════════════════════════════
# Demonstration
# ════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    # Wire handlers
    place_handler    = PlaceOrderHandler()
    pay_handler      = PayOrderHandler()
    ship_handler     = ShipOrderHandler()
    cancel_handler   = CancelOrderHandler()

    detail_handler    = GetOrderDetailHandler()
    cust_handler      = GetCustomerOrdersHandler()
    fulfill_handler   = GetFulfillmentQueueHandler()

    print("=" * 60)
    print("1. Place two orders")
    print("=" * 60)
    order_a = place_handler.handle(PlaceOrderCommand(
        customer_id="CUST-001",
        items=[
            {"sku": "LAPTOP-001", "qty": 1, "price": "1299.99"},
            {"sku": "MOUSE-001",  "qty": 2, "price":   "29.99"},
        ]
    ))
    order_b = place_handler.handle(PlaceOrderCommand(
        customer_id="CUST-001",
        items=[{"sku": "KEYBOARD-001", "qty": 1, "price": "89.99"}]
    ))

    print("\n" + "=" * 60)
    print("2. Pay order A")
    print("=" * 60)
    pay_handler.handle(PayOrderCommand(
        order_id=order_a, payment_id="PAY-xyz123",
        amount=Decimal("1359.97")
    ))

    print("\n" + "=" * 60)
    print("3. Query: Order detail (read side — no domain model access)")
    print("=" * 60)
    detail = detail_handler.handle(GetOrderDetailQuery(order_a))
    if detail:
        print(f"  Order:    {detail.order_id}")
        print(f"  Status:   {detail.status}")
        print(f"  Total:    ${detail.total}")
        print(f"  Payment:  {detail.payment_id}")
        print(f"  Items:    {len(detail.items)}")

    print("\n" + "=" * 60)
    print("4. Query: Fulfillment queue")
    print("=" * 60)
    queue = fulfill_handler.handle(GetFulfillmentQueueQuery())
    for order in queue:
        print(f"  Order {order.order_id}: {', '.join(order.item_lines)} — ${order.total}")

    print("\n" + "=" * 60)
    print("5. Ship order A — removes from fulfillment queue")
    print("=" * 60)
    ship_handler.handle(ShipOrderCommand(order_a, "TRACK-1Z999AA"))
    queue_after = fulfill_handler.handle(GetFulfillmentQueueQuery())
    print(f"  Fulfillment queue size: {len(queue_after)} (was {len(queue)})")

    print("\n" + "=" * 60)
    print("6. Cancel order B")
    print("=" * 60)
    cancel_handler.handle(CancelOrderCommand(order_b, "Customer request"))

    print("\n" + "=" * 60)
    print("7. Query: Customer order history")
    print("=" * 60)
    history = cust_handler.handle(GetCustomerOrdersQuery("CUST-001"))
    for o in history:
        print(f"  {o.order_id}: {o.status} — ${o.total} ({o.item_count} items)")

    print("\n" + "=" * 60)
    print("8. Domain rule: Cannot cancel shipped order")
    print("=" * 60)
    try:
        cancel_handler.handle(CancelOrderCommand(order_a, "Too late"))
    except ValueError as e:
        print(f"  Expected error: {e}")

Real-World Use Cases

CQRS appears wherever the complexity of reads and writes justify separate models.

E-commerce platforms: The write side processes complex order placement, inventory reservation, payment processing, and status transitions with strong domain invariants. The read side serves product search results, order history lists, recommendation feeds, and inventory dashboards — each differently shaped and independently scaled. The read side handles 100x more traffic than the write side and is served from read replicas and caches.

Banking and financial systems: Transactions are recorded on the write side with strict ACID guarantees, double-entry accounting rules, and fraud checks. The read side serves account balance views, transaction history searches, monthly statements, and risk dashboards — each a different projection of the same underlying transaction data.

Content management systems: Articles are created and edited through the write side with workflow states (draft, review, published) and version history. The read side serves the published website, search indexes, RSS feeds, and analytics dashboards — all optimized differently and cached aggressively.

Event ticketing systems: Seat reservations are handled by the write side with strict concurrency control — no two reservations for the same seat can succeed simultaneously. The read side serves seat availability maps, event listings, and booking history — shaped for each view's specific layout and filtering needs.

Healthcare records: Patient encounters, prescriptions, and lab results are recorded on the write side with strict audit requirements and domain rules. The read side serves clinician dashboards, patient portals, and billing reports — each aggregating and shaping the same data differently.

Gaming leaderboards and statistics: Game actions are recorded on the write side as events. The read side maintains denormalized leaderboard tables, player statistics, match history, and achievement progress — updated from events and cached for high-throughput reads.

Language-Specific Mistakes and Anti-Patterns

# BAD: Query handler bypasses the read model and queries the write store directly
class GetOrderDetailHandler:
    def handle(self, query: GetOrderDetailQuery) -> dict:
        order = _order_store.get(query.order_id)   # ← accessing write-side domain objects!
        if not order:
            return None
        # Now building read model from domain object in the query handler
        return {
            "id":     order.id,
            "status": order.status.value,
            "total":  str(order.total),
            "items":  [{"sku": i.sku, "qty": i.qty} for i in order.items],
        }
# Problems:
# - Read side is coupled to domain model changes — adding a field to Order breaks this query
# - Cannot optimize the read projection independently (add indexes, denormalize joins)
# - Read traffic puts load on the write model's data structures

# GOOD: Query handler reads from the read store — shaped projections only
class GetOrderDetailHandler:
    def handle(self, query: GetOrderDetailQuery) -> Optional[OrderDetailView]:
        return _order_detail_store.get(query.order_id)   # ← read store only

# BAD: Command returns the full domain object — blurs the write/read boundary
class PlaceOrderHandler:
    def handle(self, cmd: PlaceOrderCommand) -> Order:   # ← returns domain aggregate!
        order = Order.place(cmd.customer_id, cmd.items)
        _order_store[order.id] = order
        return order   # ← caller now has a mutable domain object

# Caller then uses the returned Order object as a view model — coupling read and write
result = place_handler.handle(cmd)
print(result.total)   # using domain object as DTO — wrong

# GOOD: Command returns only the new ID (or nothing); caller queries for the full view
class PlaceOrderHandler:
    def handle(self, cmd: PlaceOrderCommand) -> str:   # only the ID
        order = Order.place(cmd.customer_id, cmd.items)
        _order_store[order.id] = order
        _publish_events(order.collect_events())
        return order.id

# Caller queries explicitly for the view:
order_id = place_handler.handle(cmd)
detail   = detail_handler.handle(GetOrderDetailQuery(order_id))

Frequently Asked Questions

Q1: Do I always need two databases (one for reads, one for writes) to implement CQRS?

No. Two separate data stores are an optional optimization, not a requirement. Simple CQRS uses one database but separates the code that writes (command handlers operating on domain aggregates) from the code that reads (query handlers running direct SQL or ORM projections). This alone provides significant benefits: the domain model is clean; queries are optimized independently; the read model is shaped for its consumer.

Separate data stores (a write-optimized relational DB and a read-optimized document store, search index, or cache) are introduced when you have distinct scaling or optimization needs that a single database cannot serve. This adds eventual consistency and synchronization complexity — introduce it only when the scaling or optimization need is real.

Q2: What is the difference between CQRS and Event Sourcing? Are they the same thing?

They are different and independent patterns that are frequently combined. CQRS separates command handling from query handling — two models, two code paths, optionally two stores. Event Sourcing replaces state storage with an event log — the current state of any aggregate is derived by replaying its events.

CQRS without Event Sourcing: commands mutate state directly; the write store contains current state; events may or may not be published. Event Sourcing without CQRS: the state is stored as events, but the same model handles both reads and writes. CQRS with Event Sourcing: commands produce events that are appended to the event log; the current state is derived from the log; events are also used to update read-side projections.

The two pair naturally because Event Sourcing needs a way to produce current-state views (which CQRS read-side projections provide), and CQRS benefits from domain events as the mechanism to synchronize write and read stores.

Q3: How do I handle validation? Does it go in the command or the command handler?

Validation has two layers. Structural validation (is the command well-formed? are required fields present? are values in valid ranges?) belongs in the command handler before the domain aggregate is involved — or in a dedicated validator that the handler calls. Business/domain validation (does this state transition make sense? does this customer have permission? is inventory available?) belongs inside the domain aggregate, which enforces invariants and raises domain exceptions.

Commands themselves should be simple data carriers — validate them structurally, but do not put business logic in them. The aggregate is the guardian of business rules.

Q4: What happens when the read side is stale? How do I handle eventual consistency?

Eventual consistency is a consequence of async event propagation to the read store. Practical strategies:

Return the command result immediately: after a successful command, return the data directly from the command handler (or from the domain object before it is cleared). The caller displays this data immediately without querying the read store.

Correlation IDs and polling: return a correlation ID with the command result. The client polls the read store with the correlation ID until the projection reflects the change.

Synchronous projections for critical paths: update the read projection synchronously (in the same transaction as the write) for scenarios where immediate consistency is required. Use async projection updates for non-critical views.

Optimistic UI updates: update the UI optimistically without waiting for the projection. If the command succeeds, the UI is already showing the correct state.

Q5: Should queries ever touch the write store (domain aggregates)?

No. This is the most important discipline in CQRS. Query handlers should read exclusively from the read store — the pre-built, optimized projections. Reading from the domain aggregate in a query handler couples the read side to the write-side model, prevents independent optimization, and loads the write infrastructure with read traffic.

The only legitimate reason a query might touch the write store is if no read projection exists yet — which is a signal to create one.

Q6: How does CQRS interact with a REST API?

Commands map naturally to POST, PUT, PATCH, and DELETE endpoints. They accept input, execute side effects, and return either 201 Created with the new resource ID (for creation commands) or 204 No Content (for update/delete commands).

Queries map to GET endpoints. They accept query parameters and return view models shaped for the consumer.

One common pattern: a POST /orders endpoint sends a PlaceOrderCommand and returns 201 Created with Location: /orders/{id}. A subsequent GET /orders/{id} retrieves the order detail view from the read store. The separation is explicit in the HTTP semantics.

Q7: Is CQRS overkill for simple CRUD applications?

Yes, for most simple applications. CQRS adds structural complexity — separate command handlers, query handlers, read models, and synchronization. For a simple CRUD app where every read and write uses the same model and the domain has no complex invariants, this complexity is not justified.

The principle is: start simple, introduce CQRS when the complexity it solves exceeds the complexity it adds. Clear triggers for introduction include: read and write load differ significantly; queries require fundamentally different data shapes than the domain model; multiple teams need independent read/write development; audit and event history are required; or domain logic is complex enough to benefit from a clean write model.

Q8: Can CQRS work in a monolith, or does it require microservices?

CQRS works in any architecture. A monolith with logical CQRS separation (separate command handlers, query handlers, and read models, all in the same process) captures most of the pattern's benefits with minimal infrastructure. This is the recommended starting point.

Microservices can use CQRS where the write service and read service are deployed separately — the write service publishes events to a message bus; the read service subscribes and maintains its own read store. This requires distributed infrastructure (message bus, eventual consistency management) and is only justified by operational scaling requirements.

Q9: How do I test a CQRS system?

CQRS is highly testable by design. Command handler tests: given a command, verify that the domain aggregate transitions to the expected state and publishes the expected events. Query handler tests: given a populated read store, verify that the query returns the expected view model. Projection tests: given a domain event, verify that the projection updates the read store correctly.

The clean separation means each type of test is focused and independent. Command handler tests don't need a populated read store. Query handler tests don't need to invoke command handlers. Projection tests test only the projection logic. Integration tests can verify the full flow: send a command, process the events, query the read store.

Q10: What is a projection and how do I rebuild one?

A projection is a denormalized read model derived from domain events or state. It is shaped for a specific consumer — an order detail view, a customer history list, a fulfillment queue. Projections are the read side of CQRS.

The key advantage of event-driven projections: they can be rebuilt from scratch at any time by replaying all historical events. If you add a new field to the order detail view, rebuild the projection by replaying all OrderPlaced, OrderPaid, OrderShipped, and OrderCancelled events from the beginning. The result is the updated projection without any manual data migration.

This makes projections disposable and evolvable — they are not the source of truth (events or the write store are). They are caches of derived data that can always be regenerated.

Key Takeaways

🎯 Core Concept: CQRS separates every operation into either a Command (changes state, returns nothing or only an ID) or a Query (returns data, changes nothing). The write side owns the domain model — rich aggregates that enforce business rules, execute state transitions, and publish domain events. The read side owns query models — flat, denormalized projections shaped for each consumer's specific display needs. The two sides communicate through events: when the write side changes state, it publishes domain events that update the read-side projections. The result is a system where reads and writes are independently optimized, independently scaled, and independently evolved.

🔑 Key Benefits:

• Independent optimization: the write model is normalized for integrity; the read model is denormalized for query performance — each optimized for its purpose without compromising the other
• Independent scaling: read traffic (typically 10–100x write traffic) scales separately from write infrastructure — read replicas, caches, CDN without touching the write path
• Clean domain model: the write side is free from query convenience methods; the domain aggregate contains only business logic, not presentation concerns
• Multiple read representations: the same write-side state can project into any number of read models — order detail view, customer history, fulfillment queue — all independently shaped and cached
• Natural audit trail: domain events published by the write side capture every state change; combined with Event Sourcing, the full history is preserved
• Testability: command handlers, query handlers, and projections are independently testable — each tested in isolation with focused, small tests

🌐 Language-Specific Highlights:

• Python: Frozen @dataclass for commands, queries, and events — immutability enforced at the type level; Protocol for loose coupling between write and read sides; generator-based event replay for rebuilding projections lazily
• TypeScript: Discriminated union types (type Command = PlaceOrderCmd | CompleteTaskCmd) for type-safe command routing; readonly on all command/query/event fields; interface for read view models — structural typing enables flexible projections
• Java: record for commands, queries, events, and view models — immutable by default, concise; Optional<T> return type from query handlers — makes "not found" explicit; sealed interface + records for closed command/event hierarchies with exhaustive pattern matching
• JavaScript: Object.freeze() on all events and commands — runtime immutability; Map-based stores with clear separation; function-based handlers instead of classes — lower boilerplate; private class fields for domain state encapsulation
• C#: record types for commands, queries, and view models — value semantics, immutability; MediatR or custom mediator for command/query dispatch; ConcurrentDictionary for thread-safe in-memory stores; init-only properties in records for safe construction
• PHP: readonly classes (PHP 8.2+) for commands, queries, and events — language-enforced immutability; separate WriteRepository and ReadRepository interfaces per aggregate; arrow functions for concise projection filtering
• Go: Struct types (no classes) for commands, queries, and views — idiomatic; sync.RWMutex for concurrent read/write store access; interface-based command and query buses; function types as lightweight handler registrations
• Rust: Structs with #[derive(Debug, Clone)] for commands/queries/views; mutable vs. immutable references (&mut AppState for commands, &AppState for queries) — the type system enforces the read/write boundary; Result<T, E> for command results; Option<T> for query results
• Dart: const constructors on command/query/event objects for compile-time immutability; separate void return on command handlers vs typed return on query handlers makes the distinction syntactically clear; StreamController for async event publishing in full CQRS
• Swift: Structs (value types) for commands, queries, and view models — copied on assignment, naturally immutable; throws on command handlers for domain exceptions; class (reference type) for mutable domain aggregates; @Observable only on view models, never on domain aggregates
• Kotlin: data class for commands, queries, events, and view models — value semantics, copy() for non-destructive updates; sealed class hierarchies for closed command/event sets with exhaustive when matching; runCatching for clean command error handling; Kotlin Flows for async event streaming in full CQRS

📋 When to Use CQRS:

• Read and write traffic volumes differ significantly (reads >> writes)
• Read representations need different shapes than the domain model can efficiently serve
• Domain write logic is complex enough to benefit from a clean model unburdened by query concerns
• Multiple consumers need the same data in different shapes
• Audit trails or event history are required
• Independent team development or independent scaling of read/write sides is needed

📋 When NOT to Use CQRS:

• Simple CRUD applications with no significant domain logic
• Applications where reads and writes have the same structure and volume
• Small teams or early-stage projects where the added complexity outweighs the benefits
• Domains where strong consistency between write and read is always required and eventual consistency is unacceptable

⚠️ Anti-Patterns to Avoid:

• Query handlers accessing the write store (domain aggregates) — couples read and write
• Commands returning rich domain objects — blurs the command/query boundary
• Business logic in command objects — commands are data carriers; logic belongs in handlers and aggregates
• Polling the write store from the read side — use event-driven projection updates instead
• Single repository serving both read and write — separate WriteRepository and ReadRepository
• Domain aggregates as view models (e.g., @Observable domain classes in SwiftUI, @State domain objects in React) — domain objects are not display-optimized and expose mutation
• Neglecting projection rebuilding — projections must be rebuildable from events; design them that way from the start

🛠️ Best Practices Across Languages:

1. Immutable commands, queries, and events: Use frozen=True, readonly, record, const, or Object.freeze() — commands and events should never be mutated after creation
2. Commands return only IDs (or nothing): The new entity ID is the only acceptable return from a command handler — callers query the read store for full data
3. Query handlers read only the read store: No domain aggregate access in query handlers — query the projection, not the source of truth
4. Domain events as the synchronization mechanism: Prefer event-driven projection updates over polling, shared transactions, or direct write-store queries from the read side
5. One handler per command/query: Each command and each query maps to exactly one handler — no shared handlers, no ambiguous dispatch
6. Separate read and write repositories: UserWriteRepository (handles domain aggregates) and UserReadRepository (handles projections) — never shared
7. Make projections rebuildable: Design projections so they can be dropped and rebuilt from scratch by replaying historical events — this makes them evolvable without migration pain

💡 CQRS + Related Patterns:

• CQRS + Event Sourcing: write side stores events instead of current state; read side projects events into optimized views — the most powerful combination, highest complexity
• CQRS + Domain-Driven Design: commands target aggregates; aggregates enforce invariants; aggregates publish domain events — CQRS and DDD are natural partners
• CQRS + Mediator Pattern: a command bus and query bus route commands and queries to handlers, adding middleware (logging, validation, authorization, transactions) uniformly
• CQRS + Event-Driven Architecture: domain events flow to external consumers (other services, analytics, notifications) as well as internal projections — CQRS becomes the application-level EDA implementation
• CQRS + Saga: long-running business processes (order fulfillment, onboarding) are orchestrated by a saga that listens to domain events and issues commands — CQRS events are the Saga's input and output

This guide pairs directly with:

• Event-Driven Architecture — the infrastructure layer that transports CQRS domain events
• Event Sourcing — replacing state storage with an event log, the natural complement to CQRS
• Domain-Driven Design — the design methodology that produces the aggregates and events that CQRS separates
• Repository Pattern — the data access layer that CQRS splits into separate read and write repositories
• Mediator Pattern — the dispatcher (Command Bus / Query Bus) that routes commands and queries to handlers

Each guide includes examples in all 11 programming languages with language-specific best practices.