CQRS

Why I Stopped Using the Same Model for Reads and Writes

In the POS system, I had a reporting endpoint that listed all orders with tenant info, item counts, total amounts, and payment status — joined across four tables. The same Order model I used for creating orders, updating them, and validating business rules was being used to drive this aggregated read.

The query was slow. The model was cluttered with fields that only made sense in reporting context. Every time I changed the write model to accommodate a new business rule, it risked breaking the read path.

CQRS (Command Query Responsibility Segregation) gave me the vocabulary to separate these concerns clearly: commands change state, queries read state, and they are allowed to use completely different models optimised for their purpose.

Table of Contents

• The Core Principle

• Commands vs Queries

• Implementation Approaches

• Practical Example: Order System in the POS

• Read Model Optimisation

• CQRS Without Event Sourcing

• CQRS With Event Sourcing

• When CQRS Is Worth the Complexity

• Lessons Learned

---

The Core Principle

CQRS is based on Bertrand Meyer's Command Query Separation principle:

A method should either change the state of an object (command) or return a result (query), but not both.

CQRS applies this at the architectural level:

• Command side: Handles mutations — create, update, delete. Uses a write model optimised for business rule enforcement and consistency.

• Query side: Handles reads — lists, views, aggregations. Uses a read model optimised for the specific query's needs.

---

Commands vs Queries

Commands

A command is an intent to change state. It carries the data needed to perform the change and may be rejected if business rules are violated.

# Naming convention: imperative verb + noun
from dataclasses import dataclass

@dataclass
class CreateOrderCommand:
    tenant_id: str
    cashier_id: int
    items: list[dict]  # [{product_id, quantity, unit_price}]

@dataclass
class ApplyDiscountCommand:
    tenant_id: str
    order_id: int
    discount_percent: float
    authorised_by: int  # Manager ID

@dataclass
class VoidOrderCommand:
    tenant_id: str
    order_id: int
    reason: str
    authorised_by: int

Commands return minimal data — typically an ID or a success/failure indication. They do not return the full updated object.

Queries

A query retrieves data without changing state. It is safe to call multiple times with the same result.

# Naming convention: noun + question or list
from dataclasses import dataclass

@dataclass
class GetOrderQuery:
    tenant_id: str
    order_id: int

@dataclass
class ListOrdersQuery:
    tenant_id: str
    status: str | None = None
    cashier_id: int | None = None
    date_from: str | None = None
    date_to: str | None = None
    page: int = 1
    page_size: int = 20

@dataclass
class DailySummaryQuery:
    tenant_id: str
    date: str

---

Implementation Approaches

Level 1: Same Data Store, Separate Models

The simplest application of CQRS: keep one database, but use different code paths (command handlers vs query handlers) and different models (write model vs read DTOs).

This is what I started with in the POS system — no data replication, just separation of concerns in the code.

Level 2: Separate Read Store

Project write-side events or CDC (change data capture) into a read-optimised store — a denormalised PostgreSQL view, a dedicated read table, or a Redis cache.

Level 3: Full CQRS with Event Sourcing

The write side uses event sourcing (state is stored as a sequence of events); projections build read models from those events. The most powerful and complex approach. See Event Sourcing.

---

Practical Example: Order System in the POS

Command Side

# orders/commands/handlers.py

from dataclasses import dataclass
from .commands import CreateOrderCommand, VoidOrderCommand
from ..domain.order import Order
from ..domain.repositories import OrderRepository

class CreateOrderHandler:
    def __init__(self, repo: OrderRepository):
        self._repo = repo

    def handle(self, cmd: CreateOrderCommand) -> int:
        # Domain model validates business rules
        order = Order.create(
            tenant_id=cmd.tenant_id,
            cashier_id=cmd.cashier_id,
            items=cmd.items
        )
        saved = self._repo.save(order)
        return saved.id  # Return only the ID

class VoidOrderHandler:
    def __init__(self, repo: OrderRepository):
        self._repo = repo

    def handle(self, cmd: VoidOrderCommand) -> None:
        order = self._repo.get(cmd.tenant_id, cmd.order_id)
        if order is None:
            raise ValueError(f"Order {cmd.order_id} not found")
        order.void(reason=cmd.reason, authorised_by=cmd.authorised_by)
        self._repo.save(order)

# orders/domain/order.py — write model with business rules

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    VOIDED = "voided"

@dataclass
class OrderItem:
    product_id: int
    quantity: int
    unit_price: float

    @property
    def subtotal(self) -> float:
        return self.quantity * self.unit_price

@dataclass
class Order:
    id: int | None
    tenant_id: str
    cashier_id: int
    items: list[OrderItem]
    status: OrderStatus
    total: float
    created_at: datetime
    voided_reason: str | None = None

    @classmethod
    def create(cls, tenant_id: str, cashier_id: int, items: list[dict]) -> "Order":
        order_items = [OrderItem(**i) for i in items]
        if not order_items:
            raise ValueError("Order must have at least one item")
        total = sum(i.subtotal for i in order_items)
        return cls(
            id=None,
            tenant_id=tenant_id,
            cashier_id=cashier_id,
            items=order_items,
            status=OrderStatus.PENDING,
            total=total,
            created_at=datetime.utcnow()
        )

    def void(self, reason: str, authorised_by: int) -> None:
        if self.status == OrderStatus.VOIDED:
            raise ValueError("Order is already voided")
        self.status = OrderStatus.VOIDED
        self.voided_reason = reason

Query Side

# orders/queries/handlers.py — read model, no domain rules

from .queries import ListOrdersQuery, DailySummaryQuery
from .read_models import OrderSummary, DailySummaryReport
from sqlalchemy.orm import Session

class ListOrdersHandler:
    def __init__(self, db: Session):
        self._db = db

    def handle(self, query: ListOrdersQuery) -> list[OrderSummary]:
        # Optimised read query — joins, aggregations, pagination
        sql = """
            SELECT
                o.id,
                o.tenant_id,
                o.status,
                o.total,
                o.created_at,
                u.full_name  AS cashier_name,
                COUNT(oi.id) AS item_count
            FROM orders o
            JOIN users u ON u.id = o.cashier_id AND u.tenant_id = o.tenant_id
            JOIN order_items oi ON oi.order_id = o.id
            WHERE o.tenant_id = :tenant_id
            AND (:status IS NULL OR o.status = :status)
            AND (:cashier_id IS NULL OR o.cashier_id = :cashier_id)
            AND (:date_from IS NULL OR o.created_at >= :date_from)
            AND (:date_to IS NULL OR o.created_at <= :date_to)
            GROUP BY o.id, o.tenant_id, o.status, o.total, o.created_at, u.full_name
            ORDER BY o.created_at DESC
            LIMIT :page_size OFFSET :offset
        """
        rows = self._db.execute(sql, {
            "tenant_id": query.tenant_id,
            "status": query.status,
            "cashier_id": query.cashier_id,
            "date_from": query.date_from,
            "date_to": query.date_to,
            "page_size": query.page_size,
            "offset": (query.page - 1) * query.page_size,
        }).fetchall()

        return [OrderSummary(**dict(row)) for row in rows]

The query side uses raw SQL optimised for the specific read. It does not go through the domain model. It returns a read-specific OrderSummary DTO, not an Order domain object.

---

Read Model Optimisation

In the list orders query above, I do a JOIN with users to get the cashier name. On the write side, the Order entity does not contain the cashier name — only the cashier_id. The read model is denormalised for the specific view it supports.

For reporting queries that need aggregations (total revenue per day, top-selling products), I maintain separate materialised read tables that are updated by a background job or event projection:

-- materialised view: updated nightly or on order events
CREATE TABLE order_daily_summary (
    tenant_id    VARCHAR NOT NULL,
    date         DATE    NOT NULL,
    order_count  INT     NOT NULL,
    total_revenue DECIMAL(10,2) NOT NULL,
    avg_order_value DECIMAL(10,2) NOT NULL,
    PRIMARY KEY (tenant_id, date)
);

The DailySummaryQuery reads from this precomputed table — sub-millisecond response instead of scanning the entire orders table.

---

CQRS Without Event Sourcing

CQRS does not require event sourcing. The simplest form:

• One database, one normalised schema for writes

• Denormalised views, read-optimised queries, or a separate read table for reads

• Synchronisation between write and read model via triggers, PostgreSQL views, or a background projection job

This is the approach I use in the POS system. It provides most of the query performance benefits without the operational complexity of event sourcing.

---

CQRS With Event Sourcing

When the write side uses event sourcing, the read model is built by projecting events:

# Event projection: build read model from order events
class OrderReadModelProjector:
    def __init__(self, read_db):
        self._db = read_db

    def on_order_created(self, event: dict):
        self._db.execute(
            "INSERT INTO order_summaries (id, tenant_id, status, total, created_at) VALUES (:id, :tenant_id, :status, :total, :created_at)",
            event
        )

    def on_order_voided(self, event: dict):
        self._db.execute(
            "UPDATE order_summaries SET status = 'voided' WHERE id = :order_id",
            {"order_id": event["order_id"]}
        )

The read model can be rebuilt any time by replaying all events from the beginning — a powerful debugging and migration tool.

---

When CQRS Is Worth the Complexity

CQRS makes sense when:

• Read and write loads are asymmetric — many more reads than writes, or reads need different scaling than writes

• Read queries are complex and slow — JOINs, aggregations, and reporting views that do not fit the write model

• Domain model is rich with business rules — the write model enforces complex invariants that would make it awkward to use for reads

• Multiple read representations of the same data — a list view, a detail view, a reporting view, all optimised differently

Avoid CQRS when:

• The system is primarily CRUD with simple reads

• The team is small and introducing two code paths doubles the surface area

• There is no meaningful difference between the read and write models

---

Lessons Learned

• Start with the simplest level. Separate command and query handlers in the same database first. Add a separate read store only when query performance demands it.

• Command handlers are where domain rules live. Query handlers are where SQL craft lives. These are different skills.

• The read model is allowed to be denormalised. Redundant data in a read table is intentional, not a design flaw.

• Eventual consistency in the read model needs product sign-off. If the read model lags the write model by a second, someone needs to decide if that is acceptable.

• Naming commands as intents and queries as questions makes the codebase self-documenting. CreateOrderCommand and ListOrdersQuery say exactly what they are.