cqrs-implementation

name	cqrs-implementation
description	Implement Command Query Responsibility Segregation for scalable architectures. Use when separating read and write models, optimizing query performance, or building event-sourced systems.

CQRS Implementation

Comprehensive guide to implementing CQRS (Command Query Responsibility Segregation) patterns.

When to Use This Skill

• Separating read and write concerns
• Scaling reads independently from writes
• Building event-sourced systems
• Optimizing complex query scenarios
• Different read/write data models needed
• High-performance reporting requirements

Core Concepts

1. CQRS Architecture

                    ┌─────────────┐
                    │   Client    │
                    └──────┬──────┘
                           │
              ┌────────────┴────────────┐
              │                         │
              ▼                         ▼
       ┌─────────────┐          ┌─────────────┐
       │  Commands   │          │   Queries   │
       │    API      │          │    API      │
       └──────┬──────┘          └──────┬──────┘
              │                         │
              ▼                         ▼
       ┌─────────────┐          ┌─────────────┐
       │  Command    │          │   Query     │
       │  Handlers   │          │  Handlers   │
       └──────┬──────┘          └──────┬──────┘
              │                         │
              ▼                         ▼
       ┌─────────────┐          ┌─────────────┐
       │   Write     │─────────►│    Read     │
       │   Model     │  Events  │   Model     │
       └─────────────┘          └─────────────┘

2. Key Components

Component	Responsibility
Command	Intent to change state
Command Handler	Validates and executes commands
Event	Record of state change
Query	Request for data
Query Handler	Retrieves data from read model
Projector	Updates read model from events

Templates

Template 1: Command Infrastructure

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TypeVar, Generic, Dict, Any, Type
from datetime import datetime
import uuid

# Command base
@dataclass
class Command:
    command_id: str = None
    timestamp: datetime = None

    def __post_init__(self):
        self.command_id = self.command_id or str(uuid.uuid4())
        self.timestamp = self.timestamp or datetime.utcnow()


# Concrete commands
@dataclass
class CreateOrder(Command):
    customer_id: str
    items: list
    shipping_address: dict


@dataclass
class AddOrderItem(Command):
    order_id: str
    product_id: str
    quantity: int
    price: float


@dataclass
class CancelOrder(Command):
    order_id: 
    reason: 



T = TypeVar(, bound=Command)

 (ABC, [T]):

      () -> :
        



 :
     ():
        ._handlers: [[Command], CommandHandler] = {}

     ():
        ._handlers[command_type] = handler

      () -> :
        handler = ._handlers.get((command))
          handler:
             ValueError()
          handler.handle(command)



 (CommandHandler[CreateOrder]):
     ():
        .order_repository = order_repository
        .event_store = event_store

      () -> :
        
          command.items:
             ValueError()

        
        order = Order.create(
            customer_id=command.customer_id,
            items=command.items,
            shipping_address=command.shipping_address
        )

        
         .event_store.append_events(
            stream_id=,
            stream_type=,
            events=order.uncommitted_events
        )

         order.

Template 2: Query Infrastructure

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TypeVar, Generic, List, Optional

# Query base
@dataclass
class Query:
    pass


# Concrete queries
@dataclass
class GetOrderById(Query):
    order_id: str


@dataclass
class GetCustomerOrders(Query):
    customer_id: str
    status: Optional[str] = None
    page: int = 1
    page_size: int = 20


@dataclass
class SearchOrders(Query):
    query: str
    filters: dict = None
    sort_by: str = "created_at"
    sort_order: str = "desc"


# Query result types
@dataclass
class OrderView:
    order_id: str
    customer_id: str
    status: str
    total_amount: float
    item_count: 
    created_at: datetime
    shipped_at: [datetime] = 



 ([T]):
    items: [T]
    total: 
    page: 
    page_size: 


     () -> :
         (.total + .page_size - ) // .page_size



T = TypeVar(, bound=Query)
R = TypeVar()

 (ABC, [T, R]):

      () -> R:
        



 :
     ():
        ._handlers: [[Query], QueryHandler] = {}

     ():
        ._handlers[query_type] = handler

      () -> :
        handler = ._handlers.get((query))
          handler:
             ValueError()
          handler.handle(query)



 (QueryHandler[GetOrderById, [OrderView]]):
     ():
        .read_db = read_db

      () -> [OrderView]:
          .read_db.acquire()  conn:
            row =  conn.fetchrow(
                ,
                query.order_id
            )
             row:
                 OrderView(**(row))
             


 (QueryHandler[GetCustomerOrders, PaginatedResult[OrderView]]):
     ():
        .read_db = read_db

      () -> PaginatedResult[OrderView]:
          .read_db.acquire()  conn:
            
            where_clause = 
            params = [query.customer_id]

             query.status:
                where_clause += 
                params.append(query.status)

            
            total =  conn.fetchval(
                ,
                *params
            )

            
            offset = (query.page - ) * query.page_size
            rows =  conn.fetch(
                ,
                *params, query.page_size, offset
            )

             PaginatedResult(
                items=[OrderView(**(row))  row  rows],
                total=total,
                page=query.page,
                page_size=query.page_size
            )

Template 3: FastAPI CQRS Application

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

# Request/Response models
class CreateOrderRequest(BaseModel):
    customer_id: str
    items: List[dict]
    shipping_address: dict


class OrderResponse(BaseModel):
    order_id: str
    customer_id: str
    status: str
    total_amount: float
    item_count: int
    created_at: datetime


# Dependency injection
def get_command_bus() -> CommandBus:
    return app.state.command_bus


def get_query_bus() -> QueryBus:
    return app.state.query_bus


# Command endpoints (POST, PUT, DELETE)
@app.post("/orders", response_model=dict)
async def create_order(
    request: CreateOrderRequest,
    command_bus: CommandBus = Depends(get_command_bus)
):
    command = CreateOrder(
        customer_id=request.customer_id,
        items=request.items,
        shipping_address=request.shipping_address
    )
    order_id = await command_bus.dispatch(command)
    return {"order_id": order_id}



  ():
    command = AddOrderItem(
        order_id=order_id,
        product_id=product_id,
        quantity=quantity,
        price=price
    )
     command_bus.dispatch(command)
     {: }



  ():
    command = CancelOrder(order_id=order_id, reason=reason)
     command_bus.dispatch(command)
     {: }




  ():
    query = GetOrderById(order_id=order_id)
    result =  query_bus.dispatch(query)
      result:
         HTTPException(status_code=, detail=)
     result



  ():
    query = GetCustomerOrders(
        customer_id=customer_id,
        status=status,
        page=page,
        page_size=page_size
    )
      query_bus.dispatch(query)



  ():
    query = SearchOrders(query=q, sort_by=sort_by)
      query_bus.dispatch(query)

Template 4: Read Model Synchronization

class ReadModelSynchronizer:
    """Keeps read models in sync with events."""

    def __init__(self, event_store, read_db, projections: List[Projection]):
        self.event_store = event_store
        self.read_db = read_db
        self.projections = {p.name: p for p in projections}

    async def run(self):
        """Continuously sync read models."""
        while True:
            for name, projection in self.projections.items():
                await self._sync_projection(projection)
            await asyncio.sleep(0.1)

    async def _sync_projection(self, projection: Projection):
        checkpoint = await self._get_checkpoint(projection.name)

        events = await self.event_store.read_all(
            from_position=checkpoint,
            limit=100
        )

        for event in events:
            if event.event_type in projection.handles():
                try:
                    await projection.apply(event)
                except Exception as e:
                    
                    logger.error()
                    

             ._save_checkpoint(projection.name, event.global_position)

      ():
        
        projection = .projections[projection_name]

        
         projection.clear()

        
         ._save_checkpoint(projection_name, )

        
         :
            checkpoint =  ._get_checkpoint(projection_name)
            events =  .event_store.read_all(checkpoint, )

              events:
                

             event  events:
                 event.event_type  projection.handles():
                     projection.apply(event)

             ._save_checkpoint(
                projection_name,
                events[-].global_position
            )

Template 5: Eventual Consistency Handling

class ConsistentQueryHandler:
    """Query handler that can wait for consistency."""

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

    async def query_after_command(
        self,
        query: Query,
        expected_version: int,
        stream_id: str,
        timeout: float = 5.0
    ):
        """
        Execute query, ensuring read model is at expected version.
        Used for read-your-writes consistency.
        """
        start_time = time.time()

        while time.time() - start_time < timeout:
            # Check if read model is caught up
            projection_version = await self._get_projection_version(stream_id)

            if projection_version >= expected_version:
                return await self.execute_query(query)

            # Wait a bit and retry
            await asyncio.sleep(0.1)

        # Timeout - return stale data with warning
        return {
            "data": await self.execute_query(query),
            "_warning": "Data may be stale"
        }

    async def _get_projection_version(self, stream_id: str) -> int:
        
          .read_db.acquire()  conn:
              conn.fetchval(
                ,
                stream_id
            )  

Best Practices

Do's

• Separate command and query models - Different needs
• Use eventual consistency - Accept propagation delay
• Validate in command handlers - Before state change
• Denormalize read models - Optimize for queries
• Version your events - For schema evolution

Don'ts

• Don't query in commands - Use only for writes
• Don't couple read/write schemas - Independent evolution
• Don't over-engineer - Start simple
• Don't ignore consistency SLAs - Define acceptable lag

Resources

• CQRS Pattern
• Microsoft CQRS Guidance