CQRS — Explained with Examples
CQRS (Command Query Responsibility Segregation) splits an application into separate models for reading data and writing data to optimize each operation.
CQRS stands for Command Query Responsibility Segregation, a pattern introduced by Greg Young. It separates operations into Commands (write, mutate state) and Queries (read, return data). Each may use different models, databases, or even schemas.
Why CQRS Matters
In traditional CRUD systems, the same model handles both reads and writes — but reads and writes have different requirements. Reads need optimized projections, denormalized views, and caching. Writes need validation, transactional integrity, and event publishing. CQRS lets each side evolve independently.
Real-World Analogy
A library: writing means the librarian adding or removing books (Command) — this requires checking records, updating catalogs, and ensuring data integrity. Reading means a patron browsing the catalog (Query) — this needs fast search, filtering, and availability checks. These activities use completely different tools and workflows.
Example: CQRS in Python
from dataclasses import dataclass
from typing import Dict, List
# --- Write Side (Command) ---
@dataclass
class CreateOrderCommand:
order_id: str
product: str
quantity: int
customer: str
class OrderCommandHandler:
def __init__(self, store: Dict):
self._store = store
def handle(self, cmd: CreateOrderCommand):
# Validate, apply business rules, persist
self._store[cmd.order_id] = {
"product": cmd.product,
"quantity": cmd.quantity,
"customer": cmd.customer,
"status": "created"
}
print(f"Order {cmd.order_id} created")
# --- Read Side (Query) ---
@dataclass
class GetOrderQuery:
order_id: str
class OrderQueryHandler:
def __init__(self, read_db: Dict):
self._read_db = read_db
def handle(self, query: GetOrderQuery) -> Dict:
# Optimized read — could be denormalized, cached, etc.
return self._read_db.get(query.order_id, {})
# Usage
store = {}
handler = OrderCommandHandler(store)
handler.handle(CreateOrderCommand("123", "Widget", 5, "Alice"))
query_handler = OrderQueryHandler(store)
result = query_handler.handle(GetOrderQuery("123"))
print(result)
# Output: Order 123 created
# {'product': 'Widget', 'quantity': 5, 'customer': 'Alice', 'status': 'created'}Related Terms
Event Sourcing, CRUD, Repository Pattern, Clean Architecture
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro