CQRS in practice: separate read and write models without going full DDD

from pydantic import BaseModel
from datetime import datetime

# Commands are imperative intentions
class CreateOrderCommand(BaseModel):
    user_id: str
    items: list[dict]
    shipping_address: dict
    payment_method_id: str

class CancelOrderCommand(BaseModel):
    order_id: str
    user_id: str
    reason: str

# Command handler: validates, applies business rules, persists
class OrderCommandHandler:
    def __init__(self, db, event_bus):
        self.db = db
        self.event_bus = event_bus
    
    async def handle_create_order(self, cmd: CreateOrderCommand) -> str:
        # Business validation
        user = await self.db.users.get(cmd.user_id)
        if not user.is_active:
            raise ValueError("User account is not active")
        
        total = await self.calculate_total(cmd.items)
        
        # Write to normalized tables
        async with self.db.transaction():
            order_id = await self.db.execute(
                "INSERT INTO orders (user_id, total_cents, status) VALUES ($1, $2, 'pending') RETURNING id",
                cmd.user_id, total
            )
            
            for item in cmd.items:
                await self.db.execute(
                    "INSERT INTO order_items (order_id, product_id, qty, price_cents) VALUES ($1, $2, $3, $4)",
                    order_id, item["product_id"], item["qty"], item["price_cents"]
                )
            
            # Publish event for read model to handle
            await self.event_bus.publish("order.created", {
                "order_id": order_id,
                "user_id": cmd.user_id,
                "total_cents": total,
            })
        
        return order_id

-- Read model: denormalized for fast query
-- Built and maintained by events from the write side

CREATE TABLE order_summaries (
    order_id UUID PRIMARY KEY,
    user_id UUID NOT NULL,
    user_email TEXT NOT NULL,
    user_name TEXT NOT NULL,
    status TEXT NOT NULL,
    total_cents INTEGER NOT NULL,
    item_count INTEGER NOT NULL,
    first_item_name TEXT,
    created_at TIMESTAMPTZ NOT NULL,
    -- All the data the UI needs, pre-joined
    shipping_address JSONB
);

-- Index for common query patterns
CREATE INDEX idx_order_summaries_user ON order_summaries (user_id, created_at DESC);
CREATE INDEX idx_order_summaries_status ON order_summaries (status, created_at DESC);

class OrderReadModel:
    def __init__(self, read_db):
        self.db = read_db
    
    async def get_order(self, order_id: str) -> dict | None:
        # Simple, fast query — no joins needed
        row = await self.db.fetchrow(
            "SELECT * FROM order_summaries WHERE order_id = $1",
            order_id
        )
        return dict(row) if row else None
    
    async def get_user_orders(
        self,
        user_id: str,
        status: str | None = None,
        limit: int = 20,
    ) -> list[dict]:
        query = "SELECT * FROM order_summaries WHERE user_id = $1"
        params = [user_id]
        
        if status:
            query += " AND status = $2"
            params.append(status)
        
        query += f" ORDER BY created_at DESC LIMIT {limit}"
        
        rows = await self.db.fetch(query, *params)
        return [dict(r) for r in rows]

# Event handler: syncs read model when write model changes
class OrderReadModelProjection:
    def __init__(self, read_db, write_db):
        self.read_db = read_db
        self.write_db = write_db
    
    async def on_order_created(self, event: dict) -> None:
        # Build denormalized read model from write model data
        order = await self.write_db.fetchrow(
            """
            SELECT o.*, u.email, u.name, COUNT(oi.id) as item_count,
                   SUM(oi.price_cents * oi.qty) as total,
                   MIN(p.name) as first_item_name
            FROM orders o
            JOIN users u ON u.id = o.user_id
            JOIN order_items oi ON oi.order_id = o.id
            JOIN products p ON p.id = oi.product_id
            WHERE o.id = $1
            GROUP BY o.id, u.email, u.name
            """,
            event["order_id"]
        )
        
        await self.read_db.execute(
            """
            INSERT INTO order_summaries 
            (order_id, user_id, user_email, user_name, status, total_cents, item_count, first_item_name, created_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            ON CONFLICT (order_id) DO UPDATE SET status = EXCLUDED.status
            """,
            order["id"], order["user_id"], order["email"], order["name"],
            order["status"], order["total"], order["item_count"],
            order["first_item_name"], order["created_at"]
        )