cqrs

name	CQRS
description	Command Query Responsibility Segregation - separating read and write operations for better scalability and flexibility
category	software-development

CQRS

What I do

I separate the responsibility of updating data (commands) from reading data (queries). CQRS allows independent scaling, optimization, and evolution of read and write models. Commands change system state; queries retrieve data without side effects. This separation enables different data models, databases, and scaling strategies for reads versus writes, leading to better performance, flexibility, and maintainability in complex systems.

When to use me

Use CQRS when read and write workloads have different characteristics, when you need different data models for different use cases, or when scaling reads independently from writes is important. CQRS excels in event-driven systems, complex domain models, and systems requiring high read performance. It's valuable when the same data needs multiple optimized views. Avoid CQRS for simple CRUD applications with balanced read/write needs.

Core Concepts

• Command: Operation that changes state (create, update, delete)
• Query: Operation that reads data without modification
• Command Handler: Processes commands and emits events
• Query Handler: Retrieves data from read models
• Read Model: Optimized representation for queries
• Write Model: Domain model for business logic
• Event Sourcing: Storing state changes as events
• Synchronization: Keeping read models updated
• Projection: Building read models from events
• Eventual Consistency: Read models may lag behind writes

Code Examples

Basic CQRS Implementation

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol, Generic, TypeVar
from uuid import UUID, uuid4

T = TypeVar("T")

class Command(Protocol):
    pass

class Query(Protocol, Generic[T]):
    pass

@dataclass
class CreateUserCommand:
    email: str
    name: str

@dataclass
class UpdateUserCommand:
    user_id: UUID
    name: str

@dataclass
class DeleteUserCommand:
    user_id: UUID

@dataclass
class GetUserQuery:
    user_id: UUID

@dataclass
class ListUsersQuery:
    page: int = 1
    page_size: int = 10

@dataclass
class UserDTO:
    user_id: UUID
    email: str
    name: str
    created_at: datetime

class CommandHandler(ABC):

     () -> :
        

 (ABC, [T]):

     () -> T:
        

 :
     ():
        ._users: [UUID, ] = {}
    
     () -> :
        ._users[user[]] = user
    
     () ->  | :
         ._users.get(user_id)
    
     () -> :
        ._users.pop(user_id, )
    
     () -> []:
        users = (._users.values())
        start = (page - ) * page_size
         users[start:start + page_size]

 ():
     ():
        ._repository = repository
    
     () -> :
         (command, CreateUserCommand):
            user_id = uuid4()
            user = {
                : user_id,
                : command.email,
                : command.name,
                : datetime.utcnow()
            }
            ._repository.save(user)
        
         (command, UpdateUserCommand):
             user_id := ._repository.get(command.user_id):
                user_id[] = command.name
        
         (command, DeleteUserCommand):
            ._repository.delete(command.user_id)

 (QueryHandler[UserDTO | ]):
     ():
        ._repository = repository
    
     () -> UserDTO | :
         user := ._repository.get(query.user_id):
             UserDTO(
                user_id=user[],
                email=user[],
                name=user[],
                created_at=user[]
            )
         
    
     () -> [UserDTO]:
        users = ._repository.list_all(query.page, query.page_size)
         [
            UserDTO(
                user_id=u[],
                email=u[],
                name=u[],
                created_at=u[]
            )
             u  users
        ]

 :
     ():
        ._repository = InMemoryUserRepository()
        ._command_handler = UserCommandHandler(._repository)
        ._query_handler = UserQueryHandler(._repository)
    
     () -> :
        ._command_handler.handle(command)
    
     () -> UserDTO | :
         ._query_handler.handle(GetUserQuery(user_id))
    
     () -> [UserDTO]:
         ._query_handler.handle_list(ListUsersQuery(page, page_size))

Read Model Projections

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol

class UserEvent(Protocol):
    event_type: str
    user_id: UUID
    timestamp: datetime

@dataclass
class UserCreated:
    event_type: str = "UserCreated"
    user_id: UUID = None
    email: str = ""
    name: str = ""
    timestamp: datetime = None

@dataclass
class UserEmailChanged:
    event_type: str = "UserEmailChanged"
    user_id: UUID = None
    old_email: str = ""
    new_email: str = ""
    timestamp: datetime = None

class ReadModelProjector(ABC):
    @abstractmethod
    def project(self, event: UserEvent) -> None:
        pass

class UserDetailsReadModel:
    def __init__(self):
        self._data: [UUID, ] = {}
    
     () ->  | :
         ._data.get(user_id)
    
     () -> :
        ._data[event.user_id] = {
            : event.user_id,
            : event.email,
            : event.name,
            : event.timestamp,
            : [event.email]
        }
    
     () -> :
         user := ._data.get(event.user_id):
            user[] = event.new_email
            user[].append(event.new_email)

 :
     ():
        ._data: [, ] = {}
    
     () ->  | :
         ._data.get(email)
    
     () -> :
        ._data[event.email] = {
            : event.user_id,
            : event.email,
            : event.name
        }

 :
     ():
        ._projectors: [ReadModelProjector] = []
    
     () -> :
        ._projectors.append(projector)
    
     () -> :
         projector  ._projectors:
            projector.project(event)

 :
     ():
        ._projection_manager = projection_manager
    
     () -> :
        ._projection_manager.project(event)

Different Read Models for Different Needs

from dataclasses import dataclass
from datetime import datetime
from typing import Protocol

class OrderReadRepository(Protocol):
    pass

class OrderDetailsRepository(OrderReadRepository):
    def get_order_details(self, order_id: str) -> dict | None:
        pass
    
    def get_order_with_items(self, order_id: str) -> dict | None:
        pass

class OrderSummaryRepository(OrderReadRepository):
    def get_order_summary(self, order_id: str) -> dict | None:
        pass
    
    def get_customer_order_history(self, customer_id: str) -> list[dict]:
        pass

class SalesAnalyticsRepository(OrderReadRepository):
    def get_daily_sales(self, date: datetime) -> :
        
    
     () -> []:
        


 :
    order_id: 
    customer: 
    items: []
    shipping: 
    payment: 
    status: 


 :
    order_id: 
    customer_id: 
    total: 
    item_count: 
    status: 
    created_at: datetime


 :
    date: datetime
    total_orders: 
    total_revenue: 
    average_order_value: 

 :
     ():
        .session = session
    
     () -> OrderDetailsRepository:
         DetailsReadModel(.session)
    
     () -> OrderSummaryRepository:
         SummaryReadModel(.session)
    
     () -> SalesAnalyticsRepository:
         AnalyticsReadModel(.session)

Command Dispatching

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Type, TypeVar, Generic
from uuid import UUID

C = TypeVar("C", bound=Command)
H = TypeVar("H", bound=CommandHandler)

@dataclass
class Command:
    pass

@dataclass
class CommandResult:
    success: bool
    errors: list[str] = []
    data: dict | None = None

class CommandBus:
    def __init__(self):
        self._handlers: dict[Type[Command], CommandHandler] = {}
    
    def register(self, command_type: Type[Command], handler: CommandHandler) -> None:
        self._handlers[command_type] = handler
    
    def execute(self, command: Command) -> CommandResult:
        handler = self._handlers.get(type(command))
        if not handler:
            raise ValueError(f"No handler for {(command)}")
        
        :
            handler.handle(command)
             CommandResult(success=)
         ValidationError  e:
             CommandResult(success=, errors=[(e)])
         Exception  e:
             CommandResult(success=, errors=[])

 ():

     () -> []:
        

 ():
    

 ():
     () -> []:
        errors = []
          command.customer_id:
            errors.append()
          command.items:
            errors.append()
         command.total <= :
            errors.append()
         errors

 ():
     ():
        .customer_id = customer_id
        .items = items
        .total = total

 ():
     ():
        .repository = repository
        .event_bus = event_bus
        .validator = validator
    
     () -> :
        errors = .validator.validate(command)
         errors:
             ValidationError(errors)
        
        order_id = .repository.create(command)
        .event_bus.publish(OrderCreatedEvent(order_id))

Event Sourcing with CQRS

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Generic, TypeVar, Protocol
from uuid import uuid4, UUID

E = TypeVar("E", bound='Event')

class Event(Protocol):
    @property
    def aggregate_id(self) -> UUID:
        pass

@dataclass
class OrderEvent:
    event_id: UUID
    aggregate_id: UUID
    event_type: str
    timestamp: datetime
    version: int

class EventStore(Protocol):
    def save(self, events: list[Event]) -> None:
        pass
    
    def get_events(self, aggregate_id: UUID) -> list[Event]:
        pass

class AggregateRoot:
    def __init__(self, aggregate_id: UUID):
        self._id = aggregate_id
        self._version = 0
        ._events: [Event] = []
    
     () -> :
        ._version += 
        ._events.append(event)
    
     () -> [Event]:
         ._events.copy()
    
     () -> :
        ._events.clear()

 ():
     ():
        ().__init__(order_id)
        ._customer_id: UUID |  = 
        ._status:  = 
        ._items: [] = []
    

     () -> :
        aggregate = cls(UUID(uuid4()))
         event  events:
            aggregate._apply_event(event)
         aggregate
    
     () -> :
         (, ):
            (, )(event)
    
     () -> :
        event = OrderCreatedEvent(._, customer_id, datetime.utcnow())
        ._apply(event)
    
     () -> :
        ._customer_id = event.customer_id
        ._status = 
    
     () -> :
        event = OrderItemAddedEvent(
            ._,
            {: product_id, : quantity, : price},
            datetime.utcnow()
        )
        ._apply(event)


 ():
     ():
        ().__init__(
            event_id=uuid4(),
            aggregate_id=aggregate_id,
            event_type=,
            timestamp=timestamp,
            version=
        )
        .customer_id = customer_id


 ():
    item_data: 
    
     ():
        ().__init__(
            event_id=uuid4(),
            aggregate_id=aggregate_id,
            event_type=,
            timestamp=timestamp,
            version=
        )
        .item_data = item_data

 :
     ():
        ._event_store = event_store
    
     () -> :
        events = aggregate.get_uncommitted_events()
        ._event_store.save(events)
        aggregate.clear_uncommitted_events()
    
     () -> OrderAggregate:
        events = ._event_store.get_events(order_id)
         OrderAggregate.reconstitute(events)

Best Practices

1. Start Simple: Don't implement full CQRS unless you need it
2. Separate Models: Keep read and write models independent
3. Eventual Consistency: Accept lag between writes and reads
4. Use Events: Events make state changes explicit
5. Multiple Projections: Different views for different queries
6. Idempotency: Handle duplicate commands safely
7. Validation: Validate commands before processing
8. Sagas for Workflows: Use sagas for multi-step commands
9. Testing: Test commands and queries separately
10. Documentation: Document command and query contracts
11. Performance: Optimize hot paths in read or write models
12. Monitoring: Track command processing times and query performance