ÔöîÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÇÔöÉ

Ôöé 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: str

reason: str

Command handler base

T = TypeVar('T', bound=Command)

class CommandHandler(ABC, Generic[T]):

@abstractmethod

async def handle(self, command: T) -> Any:

pass

Command bus

class CommandBus:

def __init__(self):

self._handlers: Dict[Type[Command], CommandHandler] = {}

def register(self, command_type: Type[Command], handler: CommandHandler):

self._handlers[command_type] = handler

async def dispatch(self, command: Command) -> Any:

handler = self._handlers.get(type(command))

if not handler:

raise ValueError(f"No handler for {type(command).__name__}")

return await handler.handle(command)

Command handler implementation

class CreateOrderHandler(CommandHandler[CreateOrder]):

def __init__(self, order_repository, event_store):

self.order_repository = order_repository

self.event_store = event_store

async def handle(self, command: CreateOrder) -> str:

# Validate

if not command.items:

raise ValueError("Order must have at least one item")

# Create aggregate

order = Order.create(

customer_id=command.customer_id,

items=command.items,

shipping_address=command.shipping_address

)

# Persist events

await self.event_store.append_events(

stream_id=f"Order-{order.id}",

stream_type="Order",

events=order.uncommitted_events

)

return order.id

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: int

created_at: datetime

shipped_at: Optional[datetime] = None

@dataclass

class PaginatedResult(Generic[T]):

items: List[T]

total: int

page: int

page_size: int

@property

def total_pages(self) -> int:

return (self.total + self.page_size - 1) // self.page_size

Query handler base

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

R = TypeVar('R')

class QueryHandler(ABC, Generic[T, R]):

@abstractmethod

async def handle(self, query: T) -> R:

pass

Query bus

class QueryBus:

def __init__(self):

self._handlers: Dict[Type[Query], QueryHandler] = {}

def register(self, query_type: Type[Query], handler: QueryHandler):

self._handlers[query_type] = handler

async def dispatch(self, query: Query) -> Any:

handler = self._handlers.get(type(query))

if not handler:

raise ValueError(f"No handler for {type(query).__name__}")

return await handler.handle(query)

Query handler implementation

class GetOrderByIdHandler(QueryHandler[GetOrderById, Optional[OrderView]]):

def __init__(self, read_db):

self.read_db = read_db

async def handle(self, query: GetOrderById) -> Optional[OrderView]:

async with self.read_db.acquire() as conn:

row = await conn.fetchrow(

"""

SELECT order_id, customer_id, status, total_amount,

item_count, created_at, shipped_at

FROM order_views

WHERE order_id = $1

""",

query.order_id

)

if row:

return OrderView(**dict(row))

return None

class GetCustomerOrdersHandler(QueryHandler[GetCustomerOrders, PaginatedResult[OrderView]]):

def __init__(self, read_db):

self.read_db = read_db

async def handle(self, query: GetCustomerOrders) -> PaginatedResult[OrderView]:

async with self.read_db.acquire() as conn:

# Build query with optional status filter

where_clause = "customer_id = $1"

params = [query.customer_id]

if query.status:

where_clause += " AND status = $2"

params.append(query.status)

# Get total count

total = await conn.fetchval(

f"SELECT COUNT(*) FROM order_views WHERE {where_clause}",

*params

)

# Get paginated results

offset = (query.page - 1) * query.page_size

rows = await conn.fetch(

f"""

SELECT order_id, customer_id, status, total_amount,

item_count, created_at, shipped_at

FROM order_views

WHERE {where_clause}

ORDER BY created_at DESC

LIMIT ${len(params) + 1} OFFSET ${len(params) + 2}

""",

*params, query.page_size, offset

)

return PaginatedResult(

items=[OrderView(dict(row)) for row in rows],