checklists/ddd-checklist.md
# Domain-Driven Design Checklist
## Strategic Design
### Bounded Contexts
- [ ] Domain boundaries identified and documented
- [ ] Context map shows relationships (ACL, Shared Kernel, etc.)
- [ ] Each context has clear ownership
- [ ] Ubiquitous language defined per context
- [ ] Integration patterns chosen (events, API, shared DB)
### Ubiquitous Language
- [ ] Domain terms documented in glossary
- [ ] Code uses domain terminology (not technical jargon)
- [ ] Team (dev + domain experts) agrees on terms
- [ ] Terms are context-specific (not global)
## Tactical Design
### Entities
- [ ] Identified by unique ID (prefer UUIDv7)
- [ ] Equality based on ID, not attributes
- [ ] Contains business logic (not anemic)
- [ ] State changes through methods, not setters
- [ ] Domain events emitted for significant changes
- [ ] PostgreSQL 18: Using `gen_random_uuid_v7()` for IDs
### Value Objects
- [ ] Immutable (`frozen=True` in dataclass)
- [ ] Equality based on all attributes
- [ ] Self-validating in `__post_init__`
- [ ] No identity (no ID field)
- [ ] Operations return new instances
### Aggregates
- [ ] Aggregate root identified
- [ ] All access through aggregate root
- [ ] Invariants enforced within aggregate
- [ ] References to other aggregates by ID only
- [ ] Sized appropriately (not too large)
### Repositories
- [ ] Interface defined in domain layer (Protocol)
- [ ] Implementation in infrastructure layer
- [ ] Returns domain entities (not ORM models)
- [ ] Domain-specific query methods
- [ ] Unit of Work for transaction management
### Domain Events
- [ ] Events are immutable (frozen dataclass)
- [ ] Events named in past tense (OrderPlaced, not PlaceOrder)
- [ ] Events contain IDs, not full entities
- [ ] Collection on entity, publishing in service layer
- [ ] UUIDv7 for time-ordered event IDs
### Domain Services
- [ ] Used for cross-entity operations
- [ ] Stateless
- [ ] Named with domain verbs (not technical)
- [ ] Coordinates entities, doesn't replace their logic
## Layer Architecture
### Domain Layer
- [ ] No infrastructure dependencies
- [ ] Entities, value objects, domain events
- [ ] Repository interfaces (Protocols)
- [ ] Domain services
### Application Layer
- [ ] Use cases / application services
- [ ] Transaction management (Unit of Work)
- [ ] Event publishing
- [ ] DTO mapping
### Infrastructure Layer
- [ ] Repository implementations
- [ ] ORM models and mapping
- [ ] External service clients
- [ ] Anti-corruption layers
### Presentation Layer
- [ ] API routes/controllers
- [ ] Input validation (Pydantic)
- [ ] Response formatting
- [ ] No business logic
## Code Quality
### Naming
- [ ] Classes named with domain terms
- [ ] Methods use domain verbs
- [ ] No technical jargon in domain layer
- [ ] Consistent with ubiquitous language
### Testing
- [ ] Domain logic unit tested
- [ ] Repository implementations tested
- [ ] Application services integration tested
- [ ] Domain events verified
### Anti-Patterns Avoided
- [ ] No anemic domain model
- [ ] No business logic in controllers
- [ ] No ORM models in domain layer
- [ ] No circular dependencies between contexts
- [ ] No UUIDv4 (use UUIDv7 for time-ordering)
## PostgreSQL 18 Specifics
### UUIDv7 Usage
- [ ] `gen_random_uuid_v7()` as column default
- [ ] Python: `uuid_utils.uuid7()` for app-generated IDs
- [ ] Index on ID serves as ~created_at index
- [ ] No separate created_at index needed for sorting
### Performance
- [ ] UUIDv7 reduces index fragmentation
- [ ] Recent records clustered together
- [ ] Better cache locality for recent queries
references/bounded-contexts.md
# Bounded Contexts
## Context Map
```
┌─────────────────────────────────────────────────────────────────┐
│ E-Commerce System │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Identity │ │ Catalog │ │ Orders │ │
│ │ Context │ │ Context │ │ Context │ │
│ ├──────────────┤ ├──────────────┤ ├──────────────┤ │
│ │ • User │ │ • Product │ │ • Order │ │
│ │ • Account │ ──── │ • Category │ ──── │ • LineItem │ │
│ │ • Auth │ ACL │ • Price │ ACL │ • Payment │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ └─────────────────────┼─────────────────────┘ │
│ │ │
│ ┌──────────────────────┐ │
│ │ Shared Kernel │ │
│ │ • Money VO │ │
│ │ • Email VO │ │
│ │ • Address VO │ │
│ └──────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
ACL = Anti-Corruption Layer
```
## Context Integration Patterns
| Pattern | Use When | Direction |
|---------|----------|-----------|
| **Shared Kernel** | Teams collaborate closely | Bidirectional |
| **Customer-Supplier** | Upstream provides, downstream consumes | Unidirectional |
| **Conformist** | Must follow external model | Unidirectional |
| **Anti-Corruption Layer** | Protect domain from external changes | Inbound |
| **Open Host Service** | Provide API for many consumers | Outbound |
| **Published Language** | Standard schema (e.g., events) | Bidirectional |
## Anti-Corruption Layer
```python
# orders/infrastructure/catalog_acl.py
"""Anti-Corruption Layer for Catalog context."""
from dataclasses import dataclass
from uuid import UUID
from orders.domain.value_objects import ProductSnapshot
@dataclass
class CatalogACL:
"""Translate Catalog context to Orders context."""
def __init__(self, catalog_client: "CatalogServiceClient"):
self._catalog = catalog_client
async def get_product_snapshot(self, product_id: UUID) -> ProductSnapshot:
"""Get product data as Orders context value object.
Translates Catalog's Product entity to Orders' ProductSnapshot VO.
This protects Orders from Catalog's internal model changes.
"""
# Call external Catalog service
catalog_product = await self._catalog.get_product(product_id)
# Translate to our domain model
return ProductSnapshot(
product_id=catalog_product.id,
name=catalog_product.name,
price=Money(
amount=catalog_product.current_price,
currency=catalog_product.price_currency,
),
sku=catalog_product.sku,
# Ignore Catalog-specific fields we don't need
# like: catalog_product.category_id, .supplier_id, etc.
)
# orders/domain/value_objects.py
@dataclass(frozen=True)
class ProductSnapshot:
"""Snapshot of product at order time.
Orders context doesn't track product changes - it captures
a snapshot when the order is created.
"""
product_id: UUID
name: str
price: Money
sku: str
```
## Context Boundaries in Code
```
src/
├── identity/ # Identity Bounded Context
│ ├── domain/
│ │ ├── entities/
│ │ │ └── user.py
│ │ ├── value_objects/
│ │ │ └── email.py
│ │ └── repositories/
│ │ └── user_repository.py
│ ├── application/
│ │ └── services/
│ │ └── auth_service.py
│ └── infrastructure/
│ └── repositories/
│ └── sqlalchemy_user_repository.py
│
├── catalog/ # Catalog Bounded Context
│ ├── domain/
│ │ ├── entities/
│ │ │ ├── product.py
│ │ │ └── category.py
│ │ └── value_objects/
│ │ └── price.py
│ └── ...
│
├── orders/ # Orders Bounded Context
│ ├── domain/
│ │ ├── entities/
│ │ │ └── order.py
│ │ └── value_objects/
│ │ └── product_snapshot.py # Local copy, not shared!
│ └── infrastructure/
│ └── acl/
│ ├── catalog_acl.py # Anti-corruption layer
│ └── identity_acl.py
│
└── shared_kernel/ # Shared across contexts
└── value_objects/
├── money.py
└── address.py
```
## Cross-Context Communication
```python
# Using domain events for loose coupling
# orders/application/services/order_service.py
class OrderService:
"""Order service publishes events for other contexts."""
async def place_order(self, order: Order) -> None:
order.place()
await self._repo.update(order)
# Publish event - other contexts subscribe
await self._events.publish(OrderPlaced(
order_id=order.id,
customer_id=order.customer_id,
items=[
{"product_id": str(i.product_id), "quantity": i.quantity}
for i in order.items
],
))
# inventory/application/handlers/order_handlers.py
class OrderEventHandler:
"""Inventory context handles order events."""
async def handle_order_placed(self, event: dict) -> None:
"""Reserve inventory when order placed."""
for item in event["items"]:
await self._inventory.reserve(
product_id=UUID(item["product_id"]),
quantity=item["quantity"],
order_id=UUID(event["order_id"]),
)
```
## Context Mapping Decisions
```python
# When to use Shared Kernel
# - Both teams own the code
# - Changes require coordination
# - Strong consistency needed
# shared_kernel/value_objects/money.py
@dataclass(frozen=True)
class Money:
"""Shared by Catalog, Orders, Payments contexts."""
amount: Decimal
currency: str
# When to use ACL
# - External system you don't control
# - Legacy system with different model
# - Third-party API
# orders/infrastructure/acl/payment_gateway_acl.py
class PaymentGatewayACL:
"""Translate Stripe API to our Payment model."""
async def charge(self, payment: Payment) -> PaymentResult:
# Call Stripe with their model
stripe_charge = await self._stripe.create_charge(
amount=int(payment.amount.amount * 100), # Stripe uses cents
currency=payment.amount.currency.lower(),
source=payment.stripe_token,
)
# Translate back to our model
return PaymentResult(
success=stripe_charge.status == "succeeded",
transaction_id=stripe_charge.id,
error=stripe_charge.failure_message,
)
```
references/domain-events.md
# Domain Events
## Event Definition
```python
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import ClassVar
from uuid import UUID
from uuid_utils import uuid7 # UUIDv7 for time-ordered event IDs
@dataclass(frozen=True)
class DomainEvent:
"""Base class for domain events.
Uses UUIDv7 for time-ordered, sortable event IDs.
"""
event_id: UUID = field(default_factory=uuid7)
occurred_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
event_type: ClassVar[str]
def to_dict(self) -> dict:
"""Serialize event for publishing."""
return {
"event_id": str(self.event_id),
"event_type": self.event_type,
"occurred_at": self.occurred_at.isoformat(),
"payload": self._payload(),
}
def _payload(self) -> dict:
"""Override to provide event-specific payload."""
return {}
@dataclass(frozen=True)
class UserCreated(DomainEvent):
"""Emitted when a new user is created."""
event_type: ClassVar[str] = "user.created"
user_id: UUID = field(default_factory=uuid7)
email: str = ""
def _payload(self) -> dict:
return {"user_id": str(self.user_id), "email": self.email}
@dataclass(frozen=True)
class UserActivated(DomainEvent):
"""Emitted when user account is activated."""
event_type: ClassVar[str] = "user.activated"
user_id: UUID = field(default_factory=uuid7)
def _payload(self) -> dict:
return {"user_id": str(self.user_id)}
@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
"""Emitted when an order is placed."""
event_type: ClassVar[str] = "order.placed"
order_id: UUID = field(default_factory=uuid7)
customer_id: UUID = field(default_factory=uuid7)
total_amount: str = "0.00"
currency: str = "USD"
def _payload(self) -> dict:
return {
"order_id": str(self.order_id),
"customer_id": str(self.customer_id),
"total": {"amount": self.total_amount, "currency": self.currency},
}
```
## Entity Event Collection
```python
@dataclass
class Entity:
"""Base entity with event collection."""
_domain_events: list[DomainEvent] = field(default_factory=list, repr=False)
def add_event(self, event: DomainEvent) -> None:
"""Register domain event for later publishing."""
self._domain_events.append(event)
def collect_events(self) -> list[DomainEvent]:
"""Collect and clear pending events."""
events = self._domain_events.copy()
self._domain_events.clear()
return events
@dataclass
class Order(Entity):
"""Order with domain events."""
id: UUID = field(default_factory=uuid7)
customer_id: UUID = field(default_factory=uuid7)
status: str = "draft"
def place(self) -> None:
"""Place the order."""
if self.status != "draft":
raise ValueError("Can only place draft orders")
self.status = "placed"
self.add_event(OrderPlaced(
order_id=self.id,
customer_id=self.customer_id,
total_amount=str(self.total.amount),
currency=self.total.currency,
))
```
## Event Publisher
```python
from abc import abstractmethod
from typing import Protocol
class EventPublisher(Protocol):
"""Protocol for publishing domain events."""
@abstractmethod
async def publish(self, event: DomainEvent) -> None:
"""Publish single event."""
...
@abstractmethod
async def publish_all(self, events: list[DomainEvent]) -> None:
"""Publish multiple events."""
...
class InMemoryEventPublisher(EventPublisher):
"""In-memory publisher for testing."""
def __init__(self):
self.events: list[DomainEvent] = []
async def publish(self, event: DomainEvent) -> None:
self.events.append(event)
async def publish_all(self, events: list[DomainEvent]) -> None:
self.events.extend(events)
class RedisEventPublisher(EventPublisher):
"""Redis Streams event publisher."""
def __init__(self, redis_client, stream_name: str = "domain-events"):
self._redis = redis_client
self._stream = stream_name
async def publish(self, event: DomainEvent) -> None:
await self._redis.xadd(
self._stream,
event.to_dict(),
)
async def publish_all(self, events: list[DomainEvent]) -> None:
async with self._redis.pipeline() as pipe:
for event in events:
pipe.xadd(self._stream, event.to_dict())
await pipe.execute()
```
## Service Layer Event Publishing
```python
class OrderService:
"""Application service that publishes domain events."""
def __init__(
self,
order_repo: OrderRepository,
event_publisher: EventPublisher,
):
self._orders = order_repo
self._events = event_publisher
async def place_order(self, order_id: UUID) -> Order:
"""Place order and publish events."""
order = await self._orders.get_or_raise(order_id)
# Business logic (adds events to entity)
order.place()
# Persist changes
await self._orders.update(order)
# Publish collected events
events = order.collect_events()
await self._events.publish_all(events)
return order
```
## Event Handlers
```python
from typing import Callable, TypeVar
E = TypeVar("E", bound=DomainEvent)
class EventDispatcher:
"""Dispatch events to registered handlers."""
def __init__(self):
self._handlers: dict[str, list[Callable]] = {}
def register(
self,
event_type: str,
handler: Callable[[DomainEvent], None],
) -> None:
"""Register handler for event type."""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
async def dispatch(self, event: DomainEvent) -> None:
"""Dispatch event to all registered handlers."""
handlers = self._handlers.get(event.event_type, [])
for handler in handlers:
await handler(event)
# Handler registration
dispatcher = EventDispatcher()
async def send_welcome_email(event: UserCreated) -> None:
"""Send welcome email on user creation."""
await email_service.send_welcome(event.email)
async def update_analytics(event: UserCreated) -> None:
"""Update analytics on user creation."""
await analytics.track("user_created", {"user_id": str(event.user_id)})
dispatcher.register("user.created", send_welcome_email)
dispatcher.register("user.created", update_analytics)
```
references/entities-value-objects.md
# Entities and Value Objects
## Entity vs Value Object Decision
| Characteristic | Entity | Value Object |
|----------------|--------|--------------|
| Identity | Has unique ID | No ID, defined by attributes |
| Equality | By ID | By all attributes |
| Mutability | Mutable (state changes) | Immutable (replace whole) |
| Lifecycle | Tracked over time | Created/discarded |
| Example | User, Order, Product | Email, Money, Address |
## Entity Implementation (Python 2026)
```python
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Self
from uuid import UUID
from uuid_utils import uuid7 # pip install uuid-utils (UUIDv7 support)
from app.domain.events import DomainEvent
@dataclass
class Entity:
"""Base entity with identity and domain events.
Uses UUIDv7 for time-ordered, index-friendly IDs.
PostgreSQL 18: Use gen_random_uuid_v7() for DB-generated IDs.
"""
id: UUID = field(default_factory=uuid7)
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
_domain_events: list[DomainEvent] = field(default_factory=list, repr=False)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Entity):
return NotImplemented
return self.id == other.id
def __hash__(self) -> int:
return hash(self.id)
def add_event(self, event: DomainEvent) -> None:
"""Add domain event for later publishing."""
self._domain_events.append(event)
def collect_events(self) -> list[DomainEvent]:
"""Collect and clear domain events."""
events = self._domain_events.copy()
self._domain_events.clear()
return events
@dataclass
class User(Entity):
"""User entity with business logic."""
email: str = ""
name: str = ""
status: str = "pending"
def activate(self) -> Self:
"""Activate user account."""
if self.status == "active":
raise ValueError("User already active")
self.status = "active"
self.updated_at = datetime.now(timezone.utc)
self.add_event(UserActivated(user_id=self.id))
return self
def change_email(self, new_email: str) -> Self:
"""Change user email with validation."""
if not Email.is_valid(new_email):
raise ValueError("Invalid email format")
old_email = self.email
self.email = new_email
self.updated_at = datetime.now(timezone.utc)
self.add_event(UserEmailChanged(
user_id=self.id,
old_email=old_email,
new_email=new_email,
))
return self
```
## Value Object Implementation
```python
from dataclasses import dataclass
from decimal import Decimal
from typing import Self
import re
@dataclass(frozen=True) # Immutable!
class Email:
"""Email value object with validation."""
value: str
def __post_init__(self):
if not self.is_valid(self.value):
raise ValueError(f"Invalid email: {self.value}")
@staticmethod
def is_valid(email: str) -> bool:
pattern = r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
return bool(re.match(pattern, email))
def __str__(self) -> str:
return self.value
@dataclass(frozen=True)
class Money:
"""Money value object with currency."""
amount: Decimal
currency: str = "USD"
def __post_init__(self):
# Validate via object.__setattr__ for frozen dataclass
if self.amount < 0:
raise ValueError("Amount cannot be negative")
if len(self.currency) != 3:
raise ValueError("Currency must be 3-letter code")
def add(self, other: Self) -> Self:
"""Add money (same currency only)."""
if self.currency != other.currency:
raise ValueError("Cannot add different currencies")
return Money(self.amount + other.amount, self.currency)
def multiply(self, factor: int | Decimal) -> Self:
"""Multiply by factor."""
return Money(self.amount * Decimal(factor), self.currency)
def __str__(self) -> str:
return f"{self.currency} {self.amount:.2f}"
@dataclass(frozen=True)
class Address:
"""Address value object."""
street: str
city: str
country: str
postal_code: str
def __post_init__(self):
if not all([self.street, self.city, self.country, self.postal_code]):
raise ValueError("All address fields required")
def format_single_line(self) -> str:
return f"{self.street}, {self.city}, {self.postal_code}, {self.country}"
```
## Using Value Objects in Entities
```python
@dataclass
class Order(Entity):
"""Order entity using value objects."""
customer_email: Email = field(default_factory=lambda: Email("default@example.com"))
shipping_address: Address | None = None
total: Money = field(default_factory=lambda: Money(Decimal("0")))
items: list["OrderItem"] = field(default_factory=list)
def add_item(self, product_id: UUID, price: Money, quantity: int) -> Self:
"""Add item and recalculate total."""
item = OrderItem(
product_id=product_id,
price=price,
quantity=quantity,
)
self.items.append(item)
self.total = self._calculate_total()
return self
def _calculate_total(self) -> Money:
"""Calculate order total from items."""
total = Money(Decimal("0"))
for item in self.items:
total = total.add(item.price.multiply(item.quantity))
return total
```
## Anti-Patterns
```python
# WRONG: Value object with ID
@dataclass
class Money:
id: UUID # NO! Value objects have no identity
amount: Decimal
# WRONG: Mutable value object
@dataclass # Missing frozen=True!
class Email:
value: str
# WRONG: Entity equality by attributes
@dataclass
class User:
def __eq__(self, other):
return self.email == other.email # NO! Use ID
# WRONG: Business logic outside entity
def activate_user(user: User) -> None:
user.status = "active" # NO! Put in User.activate()
# WRONG: Using UUIDv4 in 2026
from uuid import uuid4
id: UUID = field(default_factory=uuid4) # NO! Use uuid7 for time-ordering
```
## PostgreSQL 18 UUIDv7 Integration
```sql
-- PostgreSQL 18 native UUIDv7 generation
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid_v7(),
email TEXT NOT NULL UNIQUE,
created_at TIMESTAMPTZ DEFAULT now()
);
-- UUIDv7 benefits for indexing
-- 1. Time-ordered: sequential inserts, less index fragmentation
-- 2. Sortable: ORDER BY id ≈ ORDER BY created_at
-- 3. Better cache locality: recent records clustered together
```
references/repositories.md
# Repository Pattern
## Repository Protocol (Interface)
```python
from abc import abstractmethod
from typing import Protocol, TypeVar
from uuid import UUID
from app.domain.entities import Entity
T = TypeVar("T", bound=Entity)
class Repository(Protocol[T]):
"""Generic repository protocol for domain entities."""
@abstractmethod
async def get(self, id: UUID) -> T | None:
"""Get entity by ID, returns None if not found."""
...
@abstractmethod
async def get_or_raise(self, id: UUID) -> T:
"""Get entity by ID, raises if not found."""
...
@abstractmethod
async def add(self, entity: T) -> T:
"""Add new entity to repository."""
...
@abstractmethod
async def update(self, entity: T) -> T:
"""Update existing entity."""
...
@abstractmethod
async def delete(self, id: UUID) -> None:
"""Delete entity by ID."""
...
class UserRepository(Protocol):
"""User-specific repository with domain queries."""
async def get(self, id: UUID) -> "User | None": ...
async def get_or_raise(self, id: UUID) -> "User": ...
async def add(self, user: "User") -> "User": ...
async def update(self, user: "User") -> "User": ...
async def delete(self, id: UUID) -> None: ...
# Domain-specific queries
async def find_by_email(self, email: str) -> "User | None": ...
async def find_active_users(self, limit: int = 100) -> list["User"]: ...
async def exists_by_email(self, email: str) -> bool: ...
```
## SQLAlchemy Implementation
```python
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.domain.entities import User
from app.domain.repositories import UserRepository
from app.infrastructure.models import UserModel
class SQLAlchemyUserRepository(UserRepository):
"""SQLAlchemy implementation of UserRepository."""
def __init__(self, session: AsyncSession):
self._session = session
async def get(self, id: UUID) -> User | None:
result = await self._session.get(UserModel, id)
return self._to_entity(result) if result else None
async def get_or_raise(self, id: UUID) -> User:
user = await self.get(id)
if not user:
raise UserNotFoundError(f"User {id} not found")
return user
async def add(self, user: User) -> User:
model = self._to_model(user)
self._session.add(model)
await self._session.flush()
return user
async def update(self, user: User) -> User:
model = await self._session.get(UserModel, user.id)
if not model:
raise UserNotFoundError(f"User {user.id} not found")
# Update model from entity
model.email = user.email
model.name = user.name
model.status = user.status
model.updated_at = user.updated_at
await self._session.flush()
return user
async def delete(self, id: UUID) -> None:
model = await self._session.get(UserModel, id)
if model:
await self._session.delete(model)
await self._session.flush()
async def find_by_email(self, email: str) -> User | None:
stmt = select(UserModel).where(UserModel.email == email)
result = await self._session.execute(stmt)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
async def find_active_users(self, limit: int = 100) -> list[User]:
stmt = (
select(UserModel)
.where(UserModel.status == "active")
.limit(limit)
)
result = await self._session.execute(stmt)
return [self._to_entity(m) for m in result.scalars()]
async def exists_by_email(self, email: str) -> bool:
stmt = select(UserModel.id).where(UserModel.email == email).limit(1)
result = await self._session.execute(stmt)
return result.scalar_one_or_none() is not None
def _to_entity(self, model: UserModel) -> User:
"""Map database model to domain entity."""
return User(
id=model.id,
email=model.email,
name=model.name,
status=model.status,
created_at=model.created_at,
updated_at=model.updated_at,
)
def _to_model(self, entity: User) -> UserModel:
"""Map domain entity to database model."""
return UserModel(
id=entity.id,
email=entity.email,
name=entity.name,
status=entity.status,
created_at=entity.created_at,
updated_at=entity.updated_at,
)
```
## Unit of Work Pattern
```python
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
class UnitOfWork:
"""Coordinates repositories and transaction management."""
def __init__(self, session: AsyncSession):
self._session = session
self.users = SQLAlchemyUserRepository(session)
self.orders = SQLAlchemyOrderRepository(session)
async def commit(self) -> None:
"""Commit transaction."""
await self._session.commit()
async def rollback(self) -> None:
"""Rollback transaction."""
await self._session.rollback()
@asynccontextmanager
async def unit_of_work(
session_factory,
) -> AsyncGenerator[UnitOfWork, None]:
"""Create unit of work context."""
async with session_factory() as session:
uow = UnitOfWork(session)
try:
yield uow
await uow.commit()
except Exception:
await uow.rollback()
raise
```
## Repository Best Practices
```python
# GOOD: Repository returns domain entities
async def get(self, id: UUID) -> User | None:
model = await self._session.get(UserModel, id)
return self._to_entity(model) if model else None
# BAD: Repository returns ORM models
async def get(self, id: UUID) -> UserModel | None: # Leaks infrastructure!
return await self._session.get(UserModel, id)
# GOOD: Domain-specific queries
async def find_eligible_for_discount(self) -> list[User]:
"""Find users eligible for loyalty discount."""
...
# BAD: Generic SQL queries in repository
async def find_by_query(self, query: str) -> list[User]: # Too generic!
...
# GOOD: Repository handles mapping
def _to_entity(self, model: UserModel) -> User:
return User(...)
# BAD: Caller handles mapping
user_dict = await repo.get_raw(id) # Returns dict, caller maps
```
rules/_sections.md
---
title: Domain-Driven Design Rule Categories
version: 1.0.0
---
# Rule Categories
## 1. Aggregate Boundaries (aggregates) — CRITICAL — 1 rule
Defining transactional consistency boundaries and controlling access through aggregate roots. Wrong boundaries cause data corruption and consistency violations.
- `aggregate-boundaries.md` — Aggregate root design, reference by ID, one-aggregate-per-transaction
## 2. Aggregate Invariants (invariants) — HIGH — 1 rule
Enforcing business rules within aggregate boundaries. Unenforced invariants lead to invalid domain state.
- `aggregate-invariants.md` — Business rule enforcement, specification pattern, invariant validation
## 3. Aggregate Sizing (sizing) — HIGH — 1 rule
Right-sizing aggregates for performance and consistency. Oversized aggregates cause contention; undersized ones lose consistency.
- `aggregate-sizing.md` — When to split, collection limits, eventual consistency trade-offs
rules/_template.md
---
title: "[Rule Name]"
impact: "[CRITICAL | HIGH | MEDIUM | LOW]"
impactDescription: "[What goes wrong without this rule]"
tags: []
---
## [Rule Name]
[Brief description — 1-2 sentences.]
**Incorrect:**
```
// Bad pattern
```
**Correct:**
```
// Good pattern
```
**Key rules:**
- [Rule 1]
- [Rule 2]
- [Rule 3]
Reference: [link]
rules/aggregate-boundaries.md
---
title: Define aggregate root boundaries correctly to prevent cross-transaction data corruption
impact: HIGH
impactDescription: "Wrong aggregate boundaries cause data corruption — multiple transactions modifying shared state leads to inconsistency"
tags: aggregate, root, boundary, consistency, transactional, ddd
---
## Aggregate Root Boundaries and Consistency
Aggregates define transactional consistency boundaries. The root controls all access to children and enforces the one-aggregate-per-transaction rule.
### Four Core Rules
1. **Root controls access** — External code only references aggregate root
2. **Transactional boundary** — One aggregate per transaction
3. **Reference by ID** — Never hold object references to other aggregates
4. **Invariants enforced** — Root ensures all business rules before state changes
### Correct — Aggregate Root Pattern
```python
from dataclasses import dataclass, field
from uuid import UUID
from uuid_utils import uuid7
@dataclass
class OrderAggregate:
"""Aggregate root — all access goes through here."""
id: UUID = field(default_factory=uuid7)
customer_id: UUID # Reference by ID, not Customer object!
_items: list["OrderItem"] = field(default_factory=list)
status: str = "draft"
@property
def items(self) -> tuple["OrderItem", ...]:
return tuple(self._items) # Expose immutable view
def add_item(self, product_id: UUID, quantity: int, price: "Money") -> None:
self._ensure_modifiable()
if len(self._items) >= self.MAX_ITEMS:
raise DomainError("Max items exceeded")
self._items.append(OrderItem(product_id, quantity, price))
```
### Incorrect — Cross-Aggregate References
```python
# NEVER reference aggregates by object
@dataclass
class Order:
customer: Customer # WRONG — holds object reference
# Correct: customer_id: UUID
# NEVER modify multiple aggregates in one transaction
def submit_order(order, inventory):
order.submit()
inventory.reserve(order.items) # WRONG — two aggregates in one tx
# Correct: use domain events for cross-aggregate coordination
# NEVER expose mutable collections
def items(self) -> list:
return self._items # WRONG — caller can mutate
# Correct: return tuple(self._items)
```
### Key Rules
- External code accesses children **only** through the aggregate root
- Cross-aggregate coordination uses **domain events**, not shared transactions
- Reference other aggregates by **ID**, never by object
- Expose collections as **immutable views** (tuple, frozenset)
- One aggregate = one repository = one transaction boundary
rules/aggregate-invariants.md
---
title: Enforce business invariants within aggregates to prevent invalid domain state propagation
impact: HIGH
impactDescription: "Unenforced invariants allow invalid domain state — data corruption that propagates through the system silently"
tags: invariant, business-rule, validation, specification, ddd
---
## Enforcing Business Invariants
The aggregate root is responsible for enforcing all business rules before allowing state transitions. Invariants must be checked on every mutation.
### Invariant Enforcement Pattern
```python
from dataclasses import dataclass, field
from uuid import UUID
@dataclass
class OrderAggregate:
MAX_ITEMS = 100
id: UUID
_items: list["OrderItem"] = field(default_factory=list)
status: str = "draft"
_events: list["DomainEvent"] = field(default_factory=list)
def add_item(self, product_id: UUID, quantity: int, price: "Money") -> None:
"""Add item with invariant checks."""
self._ensure_modifiable()
if len(self._items) >= self.MAX_ITEMS:
raise DomainError("Max items exceeded")
if quantity <= 0:
raise DomainError("Quantity must be positive")
self._items.append(OrderItem(product_id, quantity, price))
def submit(self) -> None:
"""Submit with business rule validation."""
self._ensure_modifiable()
if not self._items:
raise DomainError("Cannot submit empty order")
self.status = "submitted"
self._events.append(OrderSubmitted(self.id))
def _ensure_modifiable(self) -> None:
if self.status != "draft":
raise DomainError(f"Cannot modify {self.status} order")
```
### Specification Pattern for Complex Invariants
```python
from abc import ABC, abstractmethod
class Specification(ABC):
@abstractmethod
def is_satisfied_by(self, candidate) -> bool: ...
def and_(self, other: "Specification") -> "Specification":
return AndSpecification(self, other)
class OverdueOrderSpec(Specification):
def is_satisfied_by(self, order: Order) -> bool:
return (
order.status == "submitted"
and order.created_at < datetime.now() - timedelta(days=30)
)
# Usage
overdue = OverdueOrderSpec()
overdue_orders = [o for o in orders if overdue.is_satisfied_by(o)]
```
### Domain Event Collection
```python
@dataclass
class OrderAggregate:
_events: list["DomainEvent"] = field(default_factory=list)
def collect_events(self) -> list["DomainEvent"]:
"""Collect and clear events — publish AFTER persist."""
events = list(self._events)
self._events.clear()
return events
```
**Incorrect — no invariant checks, allows invalid state:**
```python
@dataclass
class OrderAggregate:
id: UUID
_items: list["OrderItem"] = field(default_factory=list)
status: str = "draft"
def add_item(self, product_id: UUID, quantity: int, price: "Money") -> None:
# No checks! Allows negative quantity, submitted order modification
self._items.append(OrderItem(product_id, quantity, price))
def submit(self) -> None:
# No check for empty order!
self.status = "submitted"
```
**Correct — enforce invariants on every mutation:**
```python
@dataclass
class OrderAggregate:
MAX_ITEMS = 100
id: UUID
_items: list["OrderItem"] = field(default_factory=list)
status: str = "draft"
def add_item(self, product_id: UUID, quantity: int, price: "Money") -> None:
self._ensure_modifiable() # Guard clause
if len(self._items) >= self.MAX_ITEMS:
raise DomainError("Max items exceeded")
if quantity <= 0:
raise DomainError("Quantity must be positive")
self._items.append(OrderItem(product_id, quantity, price))
def submit(self) -> None:
self._ensure_modifiable()
if not self._items:
raise DomainError("Cannot submit empty order")
self.status = "submitted"
def _ensure_modifiable(self) -> None:
if self.status != "draft":
raise DomainError(f"Cannot modify {self.status} order")
```
### Key Rules
- Every mutation method **checks invariants** before modifying state
- Guard clauses at the **top** of every public method
- Use the **specification pattern** for complex, reusable business rules
- Collect domain events in the aggregate, publish **after** successful persistence
- Raise `DomainError` (not generic exceptions) for invariant violations
- Status transitions follow explicit state machine rules
rules/aggregate-sizing.md
---
title: Right-size aggregates to balance lock contention against consistency guarantee requirements
impact: HIGH
impactDescription: "Oversized aggregates cause lock contention and performance degradation — undersized ones lose consistency guarantees"
tags: aggregate, sizing, performance, consistency, split, ddd
---
## Right-Sizing Aggregates
Keep aggregates small. Large aggregates cause lock contention and slow operations. Split when collections grow unbounded or when different parts change at different rates.
### Sizing Guidelines
| Signal | Action |
|--------|--------|
| < 20 children | Keep as single aggregate |
| 20-100 children | Consider splitting by access pattern |
| 100+ children | Must split — use reference by ID |
| Unbounded collection | Always split — never allow unbounded growth |
| Different change rates | Split into separate aggregates |
### Correct — Small, Focused Aggregates
```python
@dataclass
class OrderAggregate:
"""Small aggregate — bounded items list."""
id: UUID
customer_id: UUID # Reference by ID
_items: list["OrderItem"] # Bounded: max 100
MAX_ITEMS = 100
@dataclass
class CustomerAggregate:
"""Separate aggregate — customer has different lifecycle."""
id: UUID
name: str
email: str
# NO orders list here — unbounded!
```
### Incorrect — Oversized Aggregate
```python
@dataclass
class CustomerAggregate:
id: UUID
name: str
orders: list["Order"] # WRONG — unbounded growth
reviews: list["Review"] # WRONG — different change rate
notifications: list["Notification"] # WRONG — unrelated concern
```
### When to Split
1. **Unbounded collections** — If a collection can grow without limit, extract it
2. **Different change rates** — If parts of the aggregate change at different frequencies
3. **Lock contention** — If concurrent modifications frequently conflict
4. **Performance** — If loading the full aggregate is slow
### Cross-Aggregate Consistency
After splitting, use eventual consistency between aggregates:
```python
# Order aggregate publishes event
class OrderSubmitted(DomainEvent):
order_id: UUID
customer_id: UUID
# Inventory aggregate handles event (eventually consistent)
class InventoryEventHandler:
async def handle_order_submitted(self, event: OrderSubmitted) -> None:
inventory = await self.repo.get_for_order(event.order_id)
inventory.reserve_items(event.order_id)
await self.repo.save(inventory)
```
### Key Rules
- Prefer **small aggregates** (< 20 children)
- Never allow **unbounded collections** inside an aggregate
- Use **reference by ID** for cross-aggregate relationships
- Apply **eventual consistency** across aggregate boundaries via domain events
- Split by **change rate** — parts that change together stay together
- Measure **lock contention** — split if concurrent modifications conflict
scripts/entity-template.py
"""
Domain Entity Template
Generated by OrchestKit domain-driven-design skill
Uses UUIDv7 for time-ordered IDs (PostgreSQL 18 compatible).
"""
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Self
from uuid import UUID
from uuid_utils import uuid7
# =============================================================================
# Domain Events (define for this entity)
# =============================================================================
@dataclass(frozen=True)
class DomainEvent:
"""Base domain event."""
event_id: UUID = field(default_factory=uuid7)
occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC))
@dataclass(frozen=True)
class EntityCreated(DomainEvent):
"""Emitted when entity is created."""
entity_id: UUID = field(default_factory=uuid7)
@dataclass(frozen=True)
class EntityUpdated(DomainEvent):
"""Emitted when entity is updated."""
entity_id: UUID = field(default_factory=uuid7)
field_name: str = ""
old_value: str = ""
new_value: str = ""
# =============================================================================
# Base Entity
# =============================================================================
@dataclass
class Entity:
"""Base entity with UUIDv7 identity and event collection.
PostgreSQL 18: Use gen_random_uuid_v7() for DB-side ID generation.
"""
id: UUID = field(default_factory=uuid7)
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
_domain_events: list[DomainEvent] = field(default_factory=list, repr=False)
def __eq__(self, other: object) -> bool:
"""Entities are equal by identity (ID), not attributes."""
if not isinstance(other, Entity):
return NotImplemented
return self.id == other.id
def __hash__(self) -> int:
return hash(self.id)
def add_event(self, event: DomainEvent) -> None:
"""Register domain event for later publishing."""
self._domain_events.append(event)
def collect_events(self) -> list[DomainEvent]:
"""Collect and clear pending domain events."""
events = self._domain_events.copy()
self._domain_events.clear()
return events
def _touch(self) -> None:
"""Update the updated_at timestamp."""
self.updated_at = datetime.now(UTC)
# =============================================================================
# Your Entity (customize below)
# =============================================================================
@dataclass
class MyEntity(Entity):
"""TODO: Replace with your entity name and fields.
Example: User, Order, Product, etc.
"""
# TODO: Add your entity fields
name: str = ""
status: str = "active"
# email: Email = field(default_factory=lambda: Email("default@example.com"))
# -------------------------------------------------------------------------
# Factory Methods
# -------------------------------------------------------------------------
@classmethod
def create(cls, name: str) -> Self:
"""Factory method to create entity with proper initialization."""
entity = cls(name=name, status="pending")
entity.add_event(EntityCreated(entity_id=entity.id))
return entity
# -------------------------------------------------------------------------
# Business Methods (encapsulate state changes)
# -------------------------------------------------------------------------
def activate(self) -> Self:
"""Activate this entity."""
if self.status == "active":
raise ValueError("Entity already active")
old_status = self.status
self.status = "active"
self._touch()
self.add_event(EntityUpdated(
entity_id=self.id,
field_name="status",
old_value=old_status,
new_value="active",
))
return self
def deactivate(self) -> Self:
"""Deactivate this entity."""
if self.status == "inactive":
raise ValueError("Entity already inactive")
old_status = self.status
self.status = "inactive"
self._touch()
self.add_event(EntityUpdated(
entity_id=self.id,
field_name="status",
old_value=old_status,
new_value="inactive",
))
return self
def rename(self, new_name: str) -> Self:
"""Change entity name."""
if not new_name or not new_name.strip():
raise ValueError("Name cannot be empty")
old_name = self.name
self.name = new_name.strip()
self._touch()
self.add_event(EntityUpdated(
entity_id=self.id,
field_name="name",
old_value=old_name,
new_value=self.name,
))
return self
# -------------------------------------------------------------------------
# Query Methods (no side effects)
# -------------------------------------------------------------------------
@property
def is_active(self) -> bool:
"""Check if entity is active."""
return self.status == "active"
# =============================================================================
# PostgreSQL 18 Migration
# =============================================================================
"""
-- Alembic migration for PostgreSQL 18 with UUIDv7
def upgrade():
op.execute('''
CREATE TABLE my_entities (
id UUID PRIMARY KEY DEFAULT gen_random_uuid_v7(),
name TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Index on status for filtering
CREATE INDEX idx_my_entities_status ON my_entities(status);
-- UUIDv7 is time-ordered, so id index is also ~created_at index
''')
def downgrade():
op.drop_table('my_entities')
"""
scripts/repository-template.py
"""
Repository Pattern Template
Generated by OrchestKit domain-driven-design skill
SQLAlchemy 2.0 async implementation with UUIDv7 support.
"""
from abc import abstractmethod
from typing import Protocol, TypeVar
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
# =============================================================================
# Domain Entity (import from your domain layer)
# =============================================================================
# from app.domain.entities import MyEntity
class MyEntity:
"""Placeholder - replace with your actual entity."""
id: UUID
name: str
status: str
# =============================================================================
# Repository Protocol (Domain Layer)
# =============================================================================
T = TypeVar("T")
class MyEntityRepository(Protocol):
"""Repository protocol for MyEntity.
Define in domain layer - no infrastructure dependencies.
"""
@abstractmethod
async def get(self, id: UUID) -> MyEntity | None:
"""Get entity by ID, returns None if not found."""
...
@abstractmethod
async def get_or_raise(self, id: UUID) -> MyEntity:
"""Get entity by ID, raises NotFoundError if not found."""
...
@abstractmethod
async def add(self, entity: MyEntity) -> MyEntity:
"""Add new entity."""
...
@abstractmethod
async def update(self, entity: MyEntity) -> MyEntity:
"""Update existing entity."""
...
@abstractmethod
async def delete(self, id: UUID) -> None:
"""Delete entity by ID."""
...
# Domain-specific queries
@abstractmethod
async def find_by_name(self, name: str) -> MyEntity | None:
"""Find entity by name."""
...
@abstractmethod
async def find_active(self, limit: int = 100) -> list[MyEntity]:
"""Find all active entities."""
...
# =============================================================================
# SQLAlchemy Model (Infrastructure Layer)
# =============================================================================
from sqlalchemy import String
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
"""SQLAlchemy declarative base."""
pass
class MyEntityModel(Base):
"""SQLAlchemy model for MyEntity.
Uses PostgreSQL 18 gen_random_uuid_v7() for ID generation.
"""
__tablename__ = "my_entities"
# UUIDv7 generated by PostgreSQL 18
id: Mapped[UUID] = mapped_column(
PGUUID(as_uuid=True),
primary_key=True,
server_default="gen_random_uuid_v7()",
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
status: Mapped[str] = mapped_column(String(50), nullable=False, default="pending")
# =============================================================================
# Repository Implementation (Infrastructure Layer)
# =============================================================================
class EntityNotFoundError(Exception):
"""Raised when entity is not found."""
def __init__(self, entity_type: str, entity_id: UUID):
self.entity_type = entity_type
self.entity_id = entity_id
super().__init__(f"{entity_type} with id {entity_id} not found")
class SQLAlchemyMyEntityRepository(MyEntityRepository):
"""SQLAlchemy implementation of MyEntityRepository.
Handles mapping between domain entities and ORM models.
"""
def __init__(self, session: AsyncSession):
self._session = session
async def get(self, id: UUID) -> MyEntity | None:
"""Get entity by ID."""
model = await self._session.get(MyEntityModel, id)
return self._to_entity(model) if model else None
async def get_or_raise(self, id: UUID) -> MyEntity:
"""Get entity by ID, raises if not found."""
entity = await self.get(id)
if not entity:
raise EntityNotFoundError("MyEntity", id)
return entity
async def add(self, entity: MyEntity) -> MyEntity:
"""Add new entity."""
model = self._to_model(entity)
self._session.add(model)
await self._session.flush() # Get DB-generated values
return self._to_entity(model)
async def update(self, entity: MyEntity) -> MyEntity:
"""Update existing entity."""
model = await self._session.get(MyEntityModel, entity.id)
if not model:
raise EntityNotFoundError("MyEntity", entity.id)
# Update fields
model.name = entity.name
model.status = entity.status
await self._session.flush()
return entity
async def delete(self, id: UUID) -> None:
"""Delete entity by ID."""
model = await self._session.get(MyEntityModel, id)
if model:
await self._session.delete(model)
await self._session.flush()
async def find_by_name(self, name: str) -> MyEntity | None:
"""Find entity by name."""
stmt = select(MyEntityModel).where(MyEntityModel.name == name)
result = await self._session.execute(stmt)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
async def find_active(self, limit: int = 100) -> list[MyEntity]:
"""Find all active entities."""
stmt = (
select(MyEntityModel)
.where(MyEntityModel.status == "active")
.order_by(MyEntityModel.id) # UUIDv7 = time-ordered
.limit(limit)
)
result = await self._session.execute(stmt)
return [self._to_entity(m) for m in result.scalars()]
# -------------------------------------------------------------------------
# Mapping Methods
# -------------------------------------------------------------------------
def _to_entity(self, model: MyEntityModel) -> MyEntity:
"""Map ORM model to domain entity."""
entity = MyEntity()
entity.id = model.id
entity.name = model.name
entity.status = model.status
return entity
def _to_model(self, entity: MyEntity) -> MyEntityModel:
"""Map domain entity to ORM model."""
return MyEntityModel(
id=entity.id,
name=entity.name,
status=entity.status,
)
# =============================================================================
# Unit of Work Pattern
# =============================================================================
class UnitOfWork:
"""Coordinates repositories and transactions.
Usage:
async with unit_of_work(session_factory) as uow:
entity = await uow.my_entities.get(id)
entity.activate()
await uow.my_entities.update(entity)
# Auto-commits on exit
"""
def __init__(self, session: AsyncSession):
self._session = session
self.my_entities = SQLAlchemyMyEntityRepository(session)
# Add more repositories as needed
# self.orders = SQLAlchemyOrderRepository(session)
async def commit(self) -> None:
"""Commit transaction."""
await self._session.commit()
async def rollback(self) -> None:
"""Rollback transaction."""
await self._session.rollback()
# Context manager for UoW
from contextlib import asynccontextmanager
from typing import AsyncGenerator
@asynccontextmanager
async def unit_of_work(
session_factory,
) -> AsyncGenerator[UnitOfWork, None]:
"""Create unit of work with automatic commit/rollback."""
async with session_factory() as session:
uow = UnitOfWork(session)
try:
yield uow
await uow.commit()
except Exception:
await uow.rollback()
raise
scripts/value-object-template.py
"""
Value Object Templates
Generated by OrchestKit domain-driven-design skill
Value objects are immutable, compared by attributes, and self-validating.
"""
from dataclasses import dataclass
from decimal import Decimal
import re
# =============================================================================
# Email Value Object
# =============================================================================
@dataclass(frozen=True) # frozen=True makes it immutable
class Email:
"""Email address value object with validation.
Immutable: Changes create new instances.
Equality: Compared by value, not identity.
"""
value: str
def __post_init__(self) -> None:
"""Validate on construction."""
if not self._is_valid(self.value):
raise ValueError(f"Invalid email format: {self.value}")
@staticmethod
def _is_valid(email: str) -> bool:
"""Validate email format."""
pattern = r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
return bool(re.match(pattern, email))
@property
def domain(self) -> str:
"""Extract email domain."""
return self.value.split("@")[1]
@property
def local_part(self) -> str:
"""Extract local part (before @)."""
return self.value.split("@")[0]
def __str__(self) -> str:
return self.value
# =============================================================================
# Money Value Object
# =============================================================================
@dataclass(frozen=True)
class Money:
"""Money value object with currency.
Uses Decimal for precision. Immutable operations return new instances.
"""
amount: Decimal
currency: str = "USD"
def __post_init__(self) -> None:
"""Validate on construction."""
# Use object.__setattr__ for frozen dataclass validation
if not isinstance(self.amount, Decimal):
object.__setattr__(self, "amount", Decimal(str(self.amount)))
if self.amount < 0:
raise ValueError("Money amount cannot be negative")
if len(self.currency) != 3:
raise ValueError("Currency must be 3-letter ISO code")
def add(self, other: "Money") -> "Money":
"""Add money amounts (same currency only)."""
if self.currency != other.currency:
raise ValueError(
f"Cannot add {self.currency} and {other.currency}"
)
return Money(self.amount + other.amount, self.currency)
def subtract(self, other: "Money") -> "Money":
"""Subtract money amounts (same currency only)."""
if self.currency != other.currency:
raise ValueError(
f"Cannot subtract {self.currency} and {other.currency}"
)
result = self.amount - other.amount
if result < 0:
raise ValueError("Result would be negative")
return Money(result, self.currency)
def multiply(self, factor: int | float | Decimal) -> "Money":
"""Multiply by a factor."""
return Money(self.amount * Decimal(str(factor)), self.currency)
def is_zero(self) -> bool:
"""Check if amount is zero."""
return self.amount == Decimal("0")
def __str__(self) -> str:
return f"{self.currency} {self.amount:.2f}"
def __repr__(self) -> str:
return f"Money({self.amount!r}, {self.currency!r})"
# =============================================================================
# Address Value Object
# =============================================================================
@dataclass(frozen=True)
class Address:
"""Postal address value object."""
street: str
city: str
state: str
postal_code: str
country: str = "US"
def __post_init__(self) -> None:
"""Validate all fields are provided."""
required = ["street", "city", "state", "postal_code", "country"]
for field_name in required:
value = getattr(self, field_name)
if not value or not value.strip():
raise ValueError(f"Address {field_name} is required")
def format_single_line(self) -> str:
"""Format as single line."""
return f"{self.street}, {self.city}, {self.state} {self.postal_code}, {self.country}"
def format_multiline(self) -> str:
"""Format as multiple lines."""
return f"{self.street}\n{self.city}, {self.state} {self.postal_code}\n{self.country}"
# =============================================================================
# DateRange Value Object
# =============================================================================
@dataclass(frozen=True)
class DateRange:
"""Date range value object."""
from datetime import date
start: date
end: date
def __post_init__(self) -> None:
"""Validate start <= end."""
if self.start > self.end:
raise ValueError("Start date must be before or equal to end date")
@property
def days(self) -> int:
"""Number of days in range (inclusive)."""
return (self.end - self.start).days + 1
def contains(self, date: "DateRange.date") -> bool:
"""Check if date is within range."""
return self.start <= date <= self.end
def overlaps(self, other: "DateRange") -> bool:
"""Check if ranges overlap."""
return self.start <= other.end and other.start <= self.end
# =============================================================================
# Percentage Value Object
# =============================================================================
@dataclass(frozen=True)
class Percentage:
"""Percentage value object (0-100)."""
value: Decimal
def __post_init__(self) -> None:
"""Validate 0-100 range."""
if not isinstance(self.value, Decimal):
object.__setattr__(self, "value", Decimal(str(self.value)))
if not (0 <= self.value <= 100):
raise ValueError("Percentage must be between 0 and 100")
def of(self, amount: Money) -> Money:
"""Calculate percentage of money amount."""
return amount.multiply(self.value / 100)
def __str__(self) -> str:
return f"{self.value}%"
# =============================================================================
# Custom Value Object Template
# =============================================================================
@dataclass(frozen=True)
class MyValueObject:
"""TODO: Replace with your value object.
Guidelines:
1. Use frozen=True for immutability
2. Validate in __post_init__
3. Operations return new instances (never mutate)
4. Override __str__ for display
"""
# TODO: Add your fields
field1: str
field2: int = 0
def __post_init__(self) -> None:
"""Validate on construction."""
if not self.field1:
raise ValueError("field1 is required")
# Add more validation...
def with_field2(self, new_value: int) -> "MyValueObject":
"""Return new instance with updated field2."""
return MyValueObject(field1=self.field1, field2=new_value)
def __str__(self) -> str:
return f"{self.field1}:{self.field2}"
test-cases.json
{
"skill": "domain-driven-design",
"version": "1.0.0",
"testCases": [
{
"id": "negative-i-need-to-model",
"rule": "",
"query": "I need to model an Order entity with line items and a Money value object for our e-commerce domain. The order should enforce a maximum of 50 items and validate currency consistency.",
"expectedBehavior": [
"Claude creates an Order entity with identity equality (by ID, not attributes)",
"Claude creates a Money value object with frozen=True dataclass",
"Claude models line items as part of the Order aggregate",
"Claude enforces the 50-item invariant inside the aggregate root",
"Claude uses domain events for state transitions (e.g., OrderSubmitted)",
"Claude does NOT leak infrastructure concerns into the domain model"
]
},
{
"id": "edge-we-have-15-entities",
"rule": "",
"query": "We have 15 entities with complex invariants spanning multiple objects. I need to define bounded contexts for our payment, inventory, and shipping subdomains with an anti-corruption layer between them.",
"expectedBehavior": [
"Claude defines separate bounded contexts for each subdomain",
"Claude implements an Anti-Corruption Layer (ACL) at context boundaries",
"Claude references entities by ID across aggregates, not direct object references",
"Claude establishes ubiquitous language per context",
"Claude uses context mapping patterns (upstream/downstream, conformist, etc.)"
]
},
{
"id": "negative-im-building-a-simple",
"rule": "",
"query": "I'm building a simple CRUD todo app with just a Task model that has title, description, and done fields. No complex business rules.",
"expectedBehavior": [
"Claude does NOT invoke the domain-driven-design skill",
"Claude uses plain dataclasses or a simple ORM model",
"Claude avoids DDD ceremony (aggregates, repositories, domain events) for this trivial domain",
"Claude may suggest a straightforward SQLAlchemy or Django model instead"
]
},
{
"id": "aggregate-boundaries",
"rule": "aggregate-boundaries",
"query": "How do I define the aggregate root boundary for an Order with LineItems and ensure transactional consistency?",
"expectedBehavior": [
"Makes Order the aggregate root that controls all access to LineItems as children",
"Enforces one aggregate per transaction rule to maintain consistency boundaries",
"References other aggregates by ID only and never holds direct object references across boundaries",
"Ensures the root entity enforces all business rules before allowing state changes"
]
},
{
"id": "aggregate-invariants",
"rule": "aggregate-invariants",
"query": "My Order aggregate needs to enforce a maximum of 100 items and validate currency consistency. Where do these checks go?",
"expectedBehavior": [
"Places invariant enforcement inside the aggregate root on every mutation method",
"Raises domain exceptions when business rules are violated before state transition occurs",
"Validates the 100-item maximum and currency consistency within the aggregate boundary",
"Warns that unenforced invariants allow invalid domain state that propagates silently through the system"
]
},
{
"id": "aggregate-sizing",
"rule": "aggregate-sizing",
"query": "Our Order aggregate loads thousands of line items and is causing performance issues with lock contention",
"expectedBehavior": [
"Recommends splitting aggregates when child collections grow beyond 100 items",
"Uses reference by ID pattern for large collections instead of loading all children",
"Identifies different change rates as a signal to split into separate smaller aggregates",
"Never allows unbounded collections within a single aggregate boundary"
]
}
]
}