# Software Architecture Design > Implements production-grade architectural patterns including DDD tactical patterns, hexagonal architecture, CQRS, and event sourcing with typed Python implementations. - Skill: `paulpas/software-architecture-design` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/software-architecture-design` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/software-architecture-design/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/software-architecture-design --- # Modern Software Architecture Design Patterns Senior software architect implementing production-grade architectural patterns — DDD tactical patterns (Entity, Value Object, Aggregate Root, Repository, Domain Event), Hexagonal Architecture (Ports & Adapters), CQRS with separate command and query models, and Event Sourcing with append-only event stores. This skill makes the model design systems where the domain core is isolated from infrastructure, dependencies flow inward, and every architectural decision enforces clear separation of concerns through typed interfaces and explicit module boundaries. ## TL;DR Checklist - [ ] Model entities by identity and value objects by their attribute equality — never confuse them - [ ] Enforce all invariants inside Aggregate Root methods before any persistence call - [ ] Define ports as abstract protocols/interfaces BEFORE writing adapters or infrastructure code - [ ] Route dependencies inward: outer layers import inner layer abstractions, never the reverse - [ ] Separate commands (state-mutating) from queries (read-only) at the API boundary level - [ ] Store events in an append-only log — never update or delete existing events - [ ] Add snapshots for aggregates with more than 20 events to avoid replay performance degradation - [ ] Version all domain events with a schema version and provide migration strategies --- ## When to Use Use this skill when: - Designing a new system where business complexity justifies architectural patterns (not CRUD scaffolding) - Refactoring a monolithic codebase into bounded contexts with clear module boundaries - Building an event-driven system where state reconstruction from history is required - Implementing CQRS for systems with heavy read loads that differ fundamentally from write patterns - Creating domain models where invariants must never be violated (financial systems, inventory, orders) - Establishing team ownership boundaries through bounded contexts and context maps --- ## When NOT to Use Avoid this skill for: - Simple CRUD applications with minimal business logic — layered architecture is sufficient - Prototypes or MVPs where delivery speed outweighs long-term maintainability - Systems with fewer than 5 concurrent developers working on the same codebase - Real-time control systems where latency requirements make event replay impractical - When team lacks familiarity with DDD concepts — invest in domain modeling workshops first --- ## Core Workflow 1. **Identify Bounded Contexts and Subdomains** — Interview domain experts, extract the ubiquitous language, and classify subdomains as core, supporting, or generic. Each bounded context gets its own module/package boundary. **Checkpoint:** Confirm no single concept has conflicting definitions across two proposed contexts. If so, either split them into separate contexts or define a shared kernel with explicit contract rules. 2. **Model the Domain Core** — For each bounded context, identify Entities (identity-based), Value Objects (attribute-based immutability), Aggregate Roots (invariant enforcement boundaries), and Domain Services (logic spanning multiple aggregates). Define these as typed Python classes with full invariant validation in constructors. **Checkpoint:** Every aggregate root must have at least one method that enforces a cross-field invariant. If no such method exists, the model is likely missing its core business rules. 3. **Define Ports and Composition Root** — Declare all interfaces (Python `Protocol` or `abc.ABC`) that the domain core depends on for infrastructure services. Create a composition root where concrete adapters are wired together. The domain core must import zero infrastructure code. **Checkpoint:** Run `import ast` on all files in the domain package — verify no imports outside the domain package itself. 4. **Implement Write Side (Commands + Event Store)** — Build command handlers that parse input, invoke aggregate root methods to mutate state, publish domain events, and persist through repositories. For event-sourced aggregates, append events to the store rather than writing entity state directly. **Checkpoint:** Every successful command must either produce at least one domain event or explicitly return a reason for refusal (no silent failures). 5. **Implement Read Side (Query Handlers)** — Build query handlers that read from optimized projections denormalized for specific query patterns. Projections are derived from domain events and kept separate from the write model. **Checkpoint:** Verify each projection is idempotent — replaying the same event stream produces identical results. 6. **Add Cross-Cutting Concerns at Boundaries** — Place authentication, authorization, input validation, and logging in outer adapters that compose with core application services. Never embed these concerns inside domain logic. **Checkpoint:** The domain core should be testable by instantiating aggregates directly with no mock objects or external dependencies. --- ## Implementation Patterns ### Pattern 1: DDD Tactical — Entity vs Value Object Entities are defined by their unique identity and persist across state changes. Value Objects have no identity — two value objects are equal if all their attributes are equal, and they are immutable once created. Confusing these two causes subtle bugs where identity collisions or silent mutations corrupt domain state. **Key distinction rule:** If you need to answer "which one?" independently of its attributes, it's an Entity. If you only care about "what is it?", it's a Value Object. ```python from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime, timezone from decimal import Decimal from typing import Protocol # --- ENTITY: Identified by a stable UUID, not its attributes --- @dataclass(frozen=True) class Money: """Value Object: defined entirely by amount and currency. Two Money instances with the same amount and currency are interchangeable.""" amount: Decimal currency: str def __post_init__(self) -> None: if not isinstance(self.amount, Decimal): raise TypeError("Money amount must be a Decimal") if self.amount < 0: raise ValueError(f"Money amount cannot be negative: {self.amount}") if len(self.currency) != 3 or not self.currency.isupper(): raise ValueError(f"Invalid ISO 4217 currency code: '{self.currency}'") def add(self, other: Money) -> Money: """Returns a new Money — never mutates either operand.""" if self.currency != other.currency: raise ValueError( f"Cannot add different currencies: {self.currency} vs {other.currency}" ) return Money(amount=self.amount + other.amount, currency=self.currency) def subtract(self, other: Money) -> Money: diff = self.amount - other.amount if diff < 0: raise ValueError( f"Subtraction would produce negative amount: {self.amount} - {other.amount}" ) return Money(amount=diff, currency=self.currency) @dataclass(frozen=True) class OrderId: """Value Object representing a unique order identifier. Using a dedicated type instead of raw string prevents mixing up IDs across different entities (Order vs Customer vs Product). """ value: str def __post_init__(self) -> None: if not self.value or len(self.value) < 8: raise ValueError(f"OrderId must be at least 8 characters: '{self.value}'") @dataclass(frozen=True) class ProductId: """Value Object for product identification.""" value: str def __post_init__(self) -> None: if not self.value or len(self.value) < 3: raise ValueError(f"ProductId must be at least 3 characters: '{self.value}'") # --- ENTITY: Has a stable identity that persists across mutations --- @dataclass(eq=False, repr=False) class Order: """Entity: identified by order_id, not its attributes. An order with the same items but a different ID is a completely different entity.""" order_id: OrderId customer_id: OrderId # Reusing value type for consistency items: list[OrderItem] = field(default_factory=list) status: str = "PENDING" created_at: datetime = field( default_factory=lambda: datetime.now(timezone.utc) ) def add_item(self, product_id: ProductId, quantity: int, unit_price: Money) -> Order: """Add an item to the order. Validates business rules.""" if quantity <= 0: raise ValueError(f"Quantity must be positive, got {quantity}") if self.status != "PENDING": raise RuntimeError(f"Cannot add items to {self.status} order {self.order_id.value}") # Check for duplicate product in same order for existing in self.items: if existing.product_id == product_id: raise ValueError( f"Product {product_id.value} already exists in order {self.order_id.value}" ) self.items.append(OrderItem(product_id=product_id, quantity=quantity, unit_price=unit_price)) return self # Return for fluent API pattern def get_total(self) -> Money: """Calculate order total across all items.""" if not self.items: raise RuntimeError("Cannot calculate total for empty order") total = Money(amount=Decimal("0"), currency=self.items[0].unit_price.currency) for item in self.items: line_total = Money( amount=item.unit_price.amount * item.quantity, currency=item.unit_price.currency, ) total = total.add(line_total) return total def __eq__(self, other: object) -> bool: """Entities are equal by identity, not by attribute comparison.""" if not isinstance(other, Order): return False return self.order_id == other.order_id def __hash__(self) -> int: return hash(self.order_id) # --- ❌ BAD: Using raw strings for IDs mixes up entity types and provides no validation --- @dataclass(eq=False) class BadOrder: order_id: str # Could be confused with customer_id, product_id items: list[dict] # No type safety, no business logic encapsulation status: str = "pending" # String enum — any value accepted def add_item(self, product_id: str, qty: int) -> None: # No validation on quantity, no duplicate check, status check missing self.items.append({"product_id": product_id, "qty": qty}) # --- ✅ GOOD: Typed IDs (value objects), validated invariants, equality by identity --- @dataclass(eq=False) class GoodOrder: order_id: OrderId customer_id: OrderId items: list[OrderItem] = field(default_factory=list) status: str = "PENDING" def add_item(self, product_id: ProductId, quantity: int, unit_price: Money) -> None: if quantity <= 0: raise ValueError(f"Quantity must be positive, got {quantity}") if self.status != "PENDING": raise RuntimeError(f"Cannot modify {self.status} order") for existing in self.items: if existing.product_id == product_id: raise ValueError(f"Duplicate product in order: {product_id.value}") self.items.append(OrderItem(product_id=product_id, quantity=quantity, unit_price=unit_price)) ``` --- ### Pattern 2: DDD Aggregate Root — Invariant Enforcement An Aggregate Root is the entry point for all operations on a cluster of domain objects. It enforces invariants that span multiple entities within its boundary. The aggregate root controls access to internal objects and guarantees consistency before any state change is persisted. ```python from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone from decimal import Decimal from enum import Enum from typing import Protocol class OrderStatus(str, Enum): PENDING = "PENDING" CONFIRMED = "CONFIRMED" SHIPPED = "SHIPPED" DELIVERED = "DELIVERED" CANCELLED = "CANCELLED" @dataclass(frozen=True) class OrderItem: product_id: ProductId quantity: int unit_price: Money @property def line_total(self) -> Money: return Money(amount=self.unit_price.amount * self.quantity, currency=self.unit_price.currency) @dataclass(eq=False) class AggregateOrder: """Aggregate Root for the Order aggregate. All modifications to order state MUST go through this class's methods. This guarantees that business invariants are always enforced: - Status transitions follow allowed paths only - Cancellation requires CONFIRMED status (not yet shipped) - Shipping requires at least one item - Total never goes below zero """ order_id: OrderId customer_id: OrderId items: list[OrderItem] = field(default_factory=list) _status: str = "PENDING" _created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @property def status(self) -> str: return self._status # --- State Transitions (enforce allowed paths) --- def confirm(self) -> AggregateOrder: """Transition from PENDING to CONFIRMED. Invariant: Order must have items and be in PENDING status. Returns a new instance (immutable update pattern). """ if self._status != OrderStatus.PENDING.value: raise RuntimeError( f"Cannot confirm order in '{self._status}' status — " f"must be '{OrderStatus.PENDING.value}'" ) if not self.items: raise RuntimeError("Cannot confirm an empty order") from copy import deepcopy new = deepcopy(self) new._status = OrderStatus.CONFIRMED.value return new def ship(self) -> AggregateOrder: """Transition from CONFIRMED to SHIPPED. Invariant: Only confirmed orders can be shipped. """ if self._status != OrderStatus.CONFIRMED.value: raise RuntimeError( f"Cannot ship order in '{self._status}' status — " f"must be '{OrderStatus.CONFIRMED.value}'" ) if not self.items: raise RuntimeError("Cannot ship an empty order") from copy import deepcopy new = deepcopy(self) new._status = OrderStatus.SHIPPED.value return new def cancel(self) -> AggregateOrder: """Cancel the order. Invariant: Only PENDING or CONFIRMED orders can be cancelled. Cannot cancel SHIPPED or DELIVERED orders — those require returns. """ if self._status not in (OrderStatus.PENDING.value, OrderStatus.CONFIRMED.value): raise RuntimeError( f"Cannot cancel order in '{self._status}' status — " f"only PENDING or CONFIRMED can be cancelled" ) from copy import deepcopy new = deepcopy(self) new._status = OrderStatus.CANCELLED.value return new # --- Query Methods (no state mutation) --- def get_total(self) -> Money: if not self.items: raise RuntimeError("Cannot calculate total for empty order") total = Money(amount=Decimal("0"), currency=self.items[0].unit_price.currency) for item in self.items: total = total.add(item.line_total) return total def can_be_cancelled(self) -> bool: """Read-only query — does not mutate state.""" return self._status in (OrderStatus.PENDING.value, OrderStatus.CONFIRMED.value) # --- Cross-Entity Invariant Enforcement --- def remove_item(self, product_id: ProductId) -> AggregateOrder: """Remove an item from the order. Invariant enforced: After removal, remaining items must still be consistent. If no items remain, order reverts to empty state (not cancelled). """ if self._status != OrderStatus.PENDING.value: raise RuntimeError(f"Cannot remove items from {self._status} order") new_items = [item for item in self.items if item.product_id != product_id] if len(new_items) == len(self.items): raise ValueError(f"Product {product_id.value} not found in order {self.order_id.value}") from copy import deepcopy new = deepcopy(self) new.items = new_items return new # --- ❌ BAD: No invariant enforcement — any state change is allowed --- class BadAggregateOrder: def __init__(self, order_id: str): self.order_id = order_id self.status = "PENDING" # Any string accepted self.items = [] def cancel(self) -> None: # No status check — can cancel a SHIPPED or DELIVERED order! self.status = "CANCELLED" def ship(self) -> None: # No status check — can ship an empty or cancelled order! self.status = "SHIPPED" # --- ✅ GOOD: All state transitions validated through the aggregate root --- def demonstrate_correct_usage() -> None: """Example of correct aggregate root usage with invariant enforcement.""" oid = OrderId(value="ORD-2024-001") cid = OrderId(value="CUST-12345") order = AggregateOrder(order_id=oid, customer_id=cid) order = order.add_item( product_id=ProductId(value="PROD-A"), quantity=2, unit_price=Money(amount=Decimal("29.99"), currency="USD"), ) order = order.add_item( product_id=ProductId(value="PROD-B"), quantity=1, unit_price=Money(amount=Decimal("49.50"), currency="USD"), ) # Invariant: must have items to confirm order = order.confirm() # PENDING → CONFIRMED # Invariant: can cancel while confirmed cancelled = order.cancel() # CONFIRMED → CANCELLED (new instance, original unchanged) # Try shipping a cancelled order — invariant violation try: order.ship() # Raises RuntimeError except RuntimeError as e: print(f"Invariant prevented: {e}") # Total calculation across entities total = order.get_total() # $109.48 ``` --- ### Pattern 3: Hexagonal Architecture — Ports & Adapters Hexagonal Architecture (Ports and Adapters) inverts the dependency rule: inner layers never import outer layers. The domain core defines abstract interfaces (ports), and infrastructure implements those interfaces as concrete adapters. This makes the domain testable without any infrastructure dependency. **Module structure example for a complete hexagonal application:** ``` order_service/ ├── domain/ # Innermost layer — no imports from outside │ ├── __init__.py │ ├── entities.py # Entity and Value Object definitions │ ├── aggregates.py # Aggregate Root with invariant enforcement │ ├── events.py # Domain event definitions │ └── ports.py # Abstract protocol interfaces (ports) ├── application/ # Application services — orchestrates domain + ports │ ├── __init__.py │ ├── command_handlers.py # Command handlers for write operations │ ├── query_handlers.py # Query handlers for read operations │ └── service_layer.py # Use case orchestrations ├── infrastructure/ # Outer layer — imports from inner layers only │ ├── __init__.py │ ├── repositories/ # Port implementations for persistence │ │ ├── order_repo_impl.py │ │ └── event_store_impl.py │ ├── messaging/ # Adapter for external message buses │ │ └── pub_sub_adapter.py │ └── logging/ # Infrastructure concerns │ └── audit_logger.py └── composition_root.py # Wiring: create adapters, inject into application services ``` **Defining ports (abstract interfaces) in the domain layer:** ```python from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, timezone from decimal import Decimal from typing import Protocol, Sequence # --- DOMAIN LAYER: Abstract port definitions (no infrastructure imports) --- class OrderRepository(Protocol): """Port interface for order persistence. The domain layer depends on THIS ABSTRACTION, not on any specific database. Infrastructure code implements this protocol. """ @abstractmethod def save(self, order: AggregateOrder) -> None: ... @abstractmethod def find_by_id(self, order_id: OrderId) -> AggregateOrder | None: ... @abstractmethod def find_all(self, customer_id: OrderId) -> list[AggregateOrder]: ... class EventBus(Protocol): """Port interface for publishing domain events. Allows the domain to emit events without knowing about message brokers. """ @abstractmethod def publish(self, event: DomainEvent) -> None: ... @abstractmethod def subscribe(self, event_type: type[DomainEvent], handler) -> None: ... # --- Infrastructure Port Implementations (outer layer imports inner layer abstractions) --- class InMemoryOrderRepository: """Concrete implementation for development and testing. Uses in-memory dict — no database dependency. Swappable at composition root. """ def __init__(self) -> None: self._store: dict[str, AggregateOrder] = {} def save(self, order: AggregateOrder) -> None: from copy import deepcopy self._store[order.order_id.value] = deepcopy(order) def find_by_id(self, order_id: OrderId) -> AggregateOrder | None: from copy import deepcopy return deepcopy(self._store.get(order_id.value)) def find_all(self, customer_id: OrderId) -> list[AggregateOrder]: results = [ deepcopy(o) for o in self._store.values() if o.customer_id == customer_id ] return results class InMemoryEventBus: """Concrete implementation that dispatches events to registered handlers synchronously.""" def __init__(self) -> None: self._handlers: dict[type[DomainEvent], list] = {} def publish(self, event: DomainEvent) -> None: for handler in self._handlers.get(type(event), []): handler(event) def subscribe(self, event_type: type[DomainEvent], handler) -> None: from collections import defaultdict if not hasattr(self, '_handlers'): self._handlers = {} self._handlers.setdefault(event_type, []).append(handler) # --- ❌ BAD: Tight coupling — infrastructure leaks into domain --- class BadOrderService: """Violates dependency rule: domain code directly depends on SQLite.""" def __init__(self): import sqlite3 # Domain layer must never import framework! self.conn = sqlite3.connect("orders.db") def save_order(self, order_id: str) -> None: # Business logic mixed with SQL — impossible to test without DB cursor = self.conn.cursor() cursor.execute( "INSERT INTO orders (id) VALUES (?)", (order_id,) ) # --- ✅ GOOD: Domain defines protocol, infrastructure implements it --- class GoodOrderService: """Clean dependency direction: service depends on protocols, not implementations.""" def __init__( self, order_repo: OrderRepository, event_bus: EventBus, ): # Dependencies injected — can be swapped for testing or different environments self._repo = order_repo self._bus = event_bus def create_order(self, customer_id: OrderId, items: list[OrderItemData]) -> AggregateOrder: """Application service that orchestrates domain operations.""" from copy import deepcopy # Create aggregate (domain logic) oid = OrderId(value=f"ORD-{customer_id.value}-{datetime.now(timezone.utc).strftime('%s')}") order = AggregateOrder(order_id=oid, customer_id=customer_id) for item_data in items: money = Money(amount=item_data.unit_price, currency="USD") order.add_item( product_id=ProductId(value=item_data.product_id), quantity=item_data.quantity, unit_price=money, ) # Persist through the port (not directly to DB) self._repo.save(order) return order @dataclass class OrderItemData: """DTO for input data — exists in application layer, not domain.""" product_id: str quantity: int unit_price: Decimal ``` **Composition Root (wiring everything together):** ```python # composition_root.py — the single entry point that wires all dependencies from order_service.domain.ports import OrderRepository, EventBus from order_service.infrastructure.repositories.order_repo_impl import InMemoryOrderRepository from order_service.infrastructure.messaging.pub_sub_adapter import InMemoryEventBus from order_service.application.service_layer import GoodOrderService def build_application() -> GoodOrderService: """Composition root: creates all concrete implementations and injects them. In production, replace InMemory* with PostgreSQL/MQ implementations here. The domain code never changes — only the wiring differs per environment. """ # Create infrastructure adapters order_repo: OrderRepository = InMemoryOrderRepository() event_bus: EventBus = InMemoryEventBus() # Wire application services with dependencies injected return GoodOrderService(order_repo=order_repo, event_bus=event_bus) # Usage — the caller never sees concrete implementations if __name__ == "__main__": app = build_application() result = app.create_order( customer_id=OrderId(value="CUST-001"), items=[OrderItemData(product_id="PROD-A", quantity=2, unit_price=Decimal("29.99"))], ) print(f"Created order: {result.order_id.value}") ``` --- ### Pattern 4: CQRS — Command Query Responsibility Segregation CQRS separates the command model (write side with rich domain logic) from the query model (read side optimized for specific query patterns). Commands mutate state through aggregate roots; queries read from denormalized projections that are derived from events. This eliminates the ORM anti-pattern where a single model serves both reads and writes. ```python from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, timezone from decimal import Decimal from typing import Protocol # --- COMMAND SIDE: Write Model --- @dataclass(frozen=True) class PlaceOrderCommand: """Immutable command describing an intended action. Commands describe INTENT — not the resulting state change. The command handler validates and executes the intent through the domain model. """ customer_id: str items: list[OrderLineItem] coupon_code: str | None = None def __post_init__(self) -> None: if not self.items: raise ValueError("Command must include at least one line item") @dataclass(frozen=True) class OrderLineItem: product_id: str quantity: int unit_price_cents: int # Store in cents to avoid floating-point issues def __post_init__(self) -> None: if self.quantity <= 0: raise ValueError("Quantity must be positive") if self.unit_price_cents <= 0: raise ValueError("Price must be positive") @dataclass(frozen=True) class CommandResult: """Result of a command execution — success or failure reason.""" success: bool order_id: str | None = None error_message: str | None = None class ICommandHandler(Protocol): """Abstract protocol for command handlers.""" @abstractmethod async def handle(self, command: PlaceOrderCommand) -> CommandResult: ... # --- QUERY SIDE: Read Model (separate from write model) --- @dataclass(frozen=True) class OrderProjection: """Denormalized read model optimized for the order list view. Contains only the fields needed for display — no business logic, no invariants to enforce. This is a query-optimized snapshot. """ order_id: str customer_id: str items_summary: int total_cents: int status: str created_at: datetime @property def total_dollars(self) -> Decimal: return Decimal(self.total_cents) / 100 class IOrderQueryService(Protocol): """Abstract protocol for read-side queries.""" @abstractmethod async def get_order_by_id(self, order_id: str) -> OrderProjection | None: ... @abstractmethod async def get_orders_for_customer( self, customer_id: str, limit: int = 50, offset: int = 0 ) -> list[OrderProjection]: ... @abstractmethod async def search_orders( self, status_filter: str | None = None, date_from: datetime | None = None ) -> list[OrderProjection]: ... # --- Command Handler Implementation --- class PlaceOrderCommandHandler: """Handles the PlaceOrder command: validates, creates aggregate, persists.""" def __init__( self, order_repo: OrderRepository, event_bus: EventBus, ) -> None: self._repo = order_repo self._bus = event_bus async def handle(self, command: PlaceOrderCommand) -> CommandResult: # Step 1: Validate command-level constraints (application layer) total_cents = sum(item.unit_price_cents * item.quantity for item in command.items) if total_cents <= 0: return CommandResult( success=False, error_message="Order total must be positive" ) # Step 2: Create aggregate and execute domain logic (domain layer) oid = OrderId(value=f"ORD-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}-{hash(command)}") order = AggregateOrder( order_id=oid, customer_id=OrderId(value=command.customer_id), ) try: for item in command.items: money = Money(amount=Decimal(item.unit_price_cents), currency="USD") order.add_item( product_id=ProductId(value=item.product_id), quantity=item.quantity, unit_price=money, ) except (ValueError, RuntimeError) as e: return CommandResult(success=False, error_message=str(e)) # Step 3: Persist through repository port (not direct DB access) self._repo.save(order) # Step 4: Publish domain event for query model updates order_created_event = OrderCreatedEvent( order_id=order.order_id.value, customer_id=command.customer_id, total_cents=total_cents, item_count=len(command.items), ) self._bus.publish(order_created_event) return CommandResult(success=True, order_id=order.order_id.value) # --- Query Handler / Projection Builder --- class OrderProjectionBuilder: """Builds denormalized query projections from domain events. This is the read side of CQRS — separate from command processing. Projections are optimized for specific query patterns and updated asynchronously from the event bus. """ def __init__(self, projection_store: ProjectionStore) -> None: self._store = projection_store def handle_event(self, event: OrderCreatedEvent) -> None: """Handle a domain event to update the read-side projection.""" projection = OrderProjection( order_id=event.order_id, customer_id=event.customer_id, items_summary=event.item_count, total_cents=event.total_cents, status="CONFIRMED", # Derive from event type created_at=event.timestamp, ) self._store.upsert(projection) # --- ❌ BAD: Single model serving both reads and writes (the ORM anti-pattern) --- class BadCQRSModel: """Single entity used for both commands AND queries — the classic ORM anti-pattern. Problems: - Queries must join across multiple tables to assemble a view - Business logic leaks into query results as computed properties - Adding a new read-optimized view requires changing the domain model - Write validation and read optimization fight for the same structure """ def __init__(self): self.conn = sqlite3.connect("app.db") def get_order_with_details(self, order_id: str) -> dict: # N+1 query problem — joins customer, items, shipping addresses, reviews... cursor = self.conn.cursor() cursor.execute(""" SELECT o.*, c.name, c.email, i.product_name, i.quantity, i.price FROM orders o JOIN customers c ON o.customer_id = c.id LEFT JOIN order_items i ON i.order_id = o.id WHERE o.id = ? """, (order_id,)) rows = cursor.fetchall() # Manual object assembly from flat result set — error-prone and slow def place_order(self, customer_id: str, items: list[dict]) -> None: # Same model handles writes — but with completely different query needs # --- ✅ GOOD: Separate write commands and read projections --- class GoodCQRSSystem: """CQRS system with clean separation between command and query sides.""" def __init__( self, cmd_handler: ICommandHandler, query_service: IOrderQueryService, ) -> None: self._command_handler = cmd_handler self._query_service = query_service async def place_order(self, command: PlaceOrderCommand) -> CommandResult: """Write path: command goes through domain model.""" return await self._command_handler.handle(command) async def get_order(self, order_id: str) -> OrderProjection | None: """Read path: query hits optimized projection store.""" return await self._query_service.get_order_by_id(order_id) async def get_customer_orders( self, customer_id: str, limit: int = 50 ) -> list[OrderProjection]: """Read path: denormalized projection supports efficient filtering.""" return await self._query_service.get_orders_for_customer(customer_id, limit=limit) ``` --- ### Pattern 5: Event Sourcing — Append-Only Event Store Event Sourcing stores the complete history of state changes as an immutable sequence of domain events. The current state is reconstructed by replaying events from the beginning (or a snapshot). This provides full auditability, supports time-travel debugging, and naturally enables the query projections used in CQRS. ```python from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, timezone from decimal import Decimal from typing import Protocol, Sequence # --- Domain Events (immutable, versioned) --- @dataclass(frozen=True) class DomainEvent: """Base class for all domain events.""" event_id: str aggregate_id: str event_type: str timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) version: int = 1 # Event schema version for migration support def serialize(self) -> dict: """Serialize event to a dictionary for storage.""" import json data = { "event_id": self.event_id, "aggregate_id": self.aggregate_id, "event_type": self.event_type, "timestamp": self.timestamp.isoformat(), "version": self.version, } # Add event-specific fields for key, value in vars(self).items(): if key not in data: if isinstance(value, (str, int, float, bool)): data[key] = value elif isinstance(value, Decimal): data[key] = str(value) # Avoid JSON float issues return json.dumps(data) @classmethod def deserialize(cls, raw: str | dict, event_class: type[DomainEvent]) -> DomainEvent: """Deserialize from storage, handling version migrations.""" if isinstance(raw, str): import json data = json.loads(raw) else: data = raw # Version migration logic — handle schema changes over time # Version 1 events have 'quantity' field; Version 2 renamed to 'qty' if data.get("version", 1) == 1 and "qty" in data: data["quantity"] = data.pop("qty") return event_class(**{k: v for k, v in data.items() if k != "event_type"}) @dataclass(frozen=True) class OrderCreatedEvent(DomainEvent): """Event: An order was created.""" customer_id: str total_cents: int item_count: int def __post_init__(self) -> None: if not self.customer_id or len(self.customer_id) < 3: raise ValueError(f"Invalid customer_id in event: {self.customer_id}") @dataclass(frozen=True) class OrderItemAddedEvent(DomainEvent): """Event: An item was added to an existing order.""" product_id: str quantity: int unit_price_cents: int @dataclass(frozen=True) class OrderConfirmedEvent(DomainEvent): """Event: The order was confirmed (payment processed).""" payment_method: str | None = None @dataclass(frozen=True) class OrderCancelledEvent(DomainEvent): """Event: The order was cancelled by customer or system.""" reason: str = "" # --- Event Store Interface (port) --- class EventStore(Protocol): """Port interface for event persistence. Events are appended only — never updated or deleted. Each aggregate has its own event stream identified by aggregate_id. """ @abstractmethod def append(self, aggregate_id: str, events: Sequence[DomainEvent]) -> int: """Append events to the stream. Returns new stream version.""" ... @abstractmethod def load(self, aggregate_id: str) -> list[DomainEvent]: """Load all events for an aggregate from the beginning of the stream.""" ... @abstractmethod def load_with_snapshot( self, aggregate_id: str, expected_version: int ) -> tuple[list[DomainEvent], dict | None]: """Load events with optional snapshot optimization.""" ... # --- Snapshot (performance optimization) --- @dataclass(frozen=True) class Snapshot: """Snapshot captures aggregate state at a point in time. When loading an aggregate, if a snapshot exists near the current version, replay only events after the snapshot rather than from the beginning. Snapshots should be taken when event count exceeds the threshold (e.g., 20). """ aggregate_id: str snapshot_version: int state_snapshot: dict # JSON-serializable aggregate state class SnapshotStore(Protocol): @abstractmethod def save(self, snapshot: Snapshot) -> None: ... @abstractmethod def load(self, aggregate_id: str) -> Snapshot | None: ... # --- Event-Sourced Aggregate Root --- @dataclass(eq=False) class EventSourcedOrder: """Aggregate root that reconstructs state from its event history. Instead of persisting entity state directly, we append domain events to an append-only log. To get current state, we replay events through the apply_* methods. """ order_id: str = "" customer_id: str = "" items: list[OrderItemAddedEvent] = field(default_factory=list) # Event records as items status: str = "PENDING" total_cents: int = 0 _version: int = 0 # --- Reconstructor: rebuild state from events (for loading) --- def replay_events(self, events: Sequence[DomainEvent]) -> None: """Replay all events to reconstruct aggregate state.""" self._clear() for event in events: method_name = f"_apply_{event.event_type.lower().replace(' ', '_')}" if hasattr(self, method_name): getattr(self, method_name)(event) def _clear(self) -> None: """Reset to empty sta …(truncated)