CQRS Pattern — Command Query Responsibility Segregation Guide

In this tutorial, you'll learn about CQRS Pattern. We cover key concepts, practical examples, and best practices.

CQRS (Command Query Responsibility Segregation) separates read and write operations into different models — commands handle mutations with validation, queries return data through optimized read-optimized views — enabling independent scaling and optimization of each side.

Why CQRS Matters

Most applications have asymmetric read and write workloads. A social media feed might serve 10,000 reads per second but only 100 writes. In a traditional CRUD approach, the same normalized database model handles both — forcing complex joins for every read. CQRS lets you optimize writes for consistency and reads for speed. Uber uses CQRS to separate trip booking (writes) from the rider-facing ETA and driver location views (reads). Durga Antivirus Pro's scan orchestration uses CQRS — scan commands go through a strict validation pipeline while scan results are served from a denormalized read cache.

Plain-Language Explanation

Imagine a library. The command side is the checkout desk — a librarian validates your library card, checks if the book is available, and updates the catalog. Strict rules apply. The query side is the card catalog — you search by title, author, or genre and instantly see which books are available and where they're shelved. These are different interfaces for different purposes, even though they work with the same underlying collection.

graph LR
    Client -->|Command: PlaceOrder| Command[Command Model]
    Client -->|Query: GetOrderHistory| Query[Query Model]
    Command --> Validator[Validation & Authorization]
    Validator --> EventStore[(Event Store)]
    EventStore --> Projector[Projection / Materializer]
    Projector --> ReadDB[(Read-Optimized DB)]
    Query --> ReadDB
    EventStore --> Queue[Message Queue]
    Queue --> Other[Other Services]
    style Command fill:#e74c3c,color:#fff
    style Query fill:#27ae60,color:#fff
    style ReadDB fill:#3498db,color:#fff
    style EventStore fill:#f39c12,color:#fff

Command Side Implementation

Commands validate input and produce events:

from pydantic import BaseModel, Field
from uuid import uuid4

class PlaceOrderCommand(BaseModel):
    user_id: str
    items: list[dict]
    shipping_address: str

class OrderCommandHandler:
    def __init__(self, event_store):
        self.event_store = event_store

    def handle(self, cmd: PlaceOrderCommand) -> dict:
        order_id = uuid4().hex[:12]
        total = sum(item["price"] * item["qty"] for item in cmd.items)
        event = {
            "type": "OrderPlaced",
            "order_id": order_id,
            "user_id": cmd.user_id,
            "items": cmd.items,
            "total": total,
            "shipping_address": cmd.shipping_address,
        }
        self.event_store.append(event)
        return {"order_id": order_id, "status": "pending"}

Query Side Implementation

Queries read from a materialized view optimized for the use case:

class OrderQueryHandler:
    def __init__(self, read_db):
        self.read_db = read_db

    def get_order_summary(self, user_id: str) -> list:
        return self.read_db.query(
            """SELECT order_id, total, status, created_at
               FROM order_summary
               WHERE user_id = :uid
               ORDER BY created_at DESC
               LIMIT 20""",
            {"uid": user_id},
        )

    def get_order_detail(self, order_id: str) -> dict:
        return self.read_db.query(
            """SELECT * FROM order_details WHERE order_id = :oid""",
            {"oid": order_id},
        )

Projection / Materialization

A projection service builds read models from the event stream:

class OrderProjector:
    def __init__(self, read_db, event_store):
        self.read_db = read_db
        self.event_store = event_store

    def rebuild(self):
        self.read_db.execute("TRUNCATE order_summary")
        for event in self.event_store.get_all():
            self.apply(event)

    def apply(self, event: dict):
        if event["type"] == "OrderPlaced":
            self.read_db.execute(
                """INSERT INTO order_summary
                   (order_id, user_id, total, status)
                   VALUES (:oid, :uid, :total, 'pending')""",
                event,
            )

Common Mistakes

1. CQRS without reason: Adding CQRS to a simple CRUD app adds complexity without benefit. Use it only when read and write patterns differ significantly.

2. Eventual consistency surprises: Commands are immediately visible on the command side but may take seconds to appear on the query side. Design your UI accordingly.

3. Shared schema: If command and query models use the same database schema, you're not doing CQRS. They should be independently optimized.

4. Complex projections: Every materialized view must be maintained. Too many projections create a maintenance burden. Start with one or two.

5. Ignoring the event store: Without event sourcing, CQRS loses its main advantage — rebuilding read models from history becomes impossible.

Practice Questions

1. What is the main benefit of CQRS? Independent optimization of read and write models. Writes can be validated strictly in a normalized model while reads use denormalized, pre-joined views.

2. How does CQRS relate to event sourcing? They pair naturally: commands produce events, and read models are projections of the event stream. CQRS works without event sourcing, but loses rebuild capability.

3. When should you NOT use CQRS? For simple CRUD applications where reads and writes access the same data in the same shape. The complexity isn't justified.

4. What is a materialized view? A pre-computed, denormalized representation of data optimized for specific query patterns, rebuilt from events as they arrive.

5. How do you handle eventual consistency in CQRS? Use UI patterns like optimistic UI (show the expected result immediately) and stale-data indicators. Notify users when data syncs.

Mini Project

Build a simple CQRS system for a todo app:

from uuid import uuid4

events = []
read_db = {"todos": []}

class TodoCommand:
    def create_todo(self, title: str):
        event = {"type": "TodoCreated", "id": uuid4().hex[:8], "title": title}
        events.append(event)
        return event["id"]

    def complete_todo(self, todo_id: str):
        events.append({"type": "TodoCompleted", "id": todo_id})

def rebuild_read_model():
    todos = {}
    for e in events:
        if e["type"] == "TodoCreated":
            todos[e["id"]] = {"id": e["id"], "title": e["title"], "done": False}
        elif e["type"] == "TodoCompleted":
            if e["id"] in todos:
                todos[e["id"]]["done"] = True
    read_db["todos"] = list(todos.values())

cmd = TodoCommand()
cmd.create_todo("Learn CQRS")
cmd.create_todo("Build a project")
cmd.complete_todo(events[0]["id"])
rebuild_read_model()
for t in read_db["todos"]:
    print(f"{t['title']} - {'done' if t['done'] else 'pending'}")

Expected output:

Learn CQRS - done
Build a project - pending

Cross-References

• Event Sourcing
• Saga Pattern
• Microservices Patterns
• Event-Driven Architecture
• API Design