ddd-tactical

DDD Tactical Patterns

Implement Domain-Driven Design tactical patterns including Entities, Value Objects, Aggregates, Domain Events, Repositories, and Domain Services.

When to Use

• When modeling domain entities with rich behavior
• When creating value objects for validated types
• When defining aggregate boundaries and invariants
• When implementing domain events for cross-aggregate communication
• When business logic spans multiple entities (Domain Services)

Available Operations

1. Create Value Objects with validation
2. Create Entities with identity and behavior
3. Define Aggregates with invariant protection
4. Implement Domain Events
5. Create Domain Services for cross-entity logic

Multi-Step Workflow

Step 1: Identify Value Objects

Value Objects are immutable, compared by value, and contain self-validation.

@dataclass(frozen=True)
class Email:
    value: str

    def __post_init__(self):
        if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', self.value):
            raise InvalidEmailException(self.value)

@dataclass(frozen=True)
class ComplianceScore:
    value: float

    def __post_init__(self):
        if not 0 <= self.value <= 100:
            raise ValueError("Score must be between 0 and 100")

    def is_passing(self) -> bool:
        return self.value >= 80.0

    def __str__(self) -> str:
        return f"{self.value:.1f}%"

Step 2: Create Entities

Entities have unique identity, are mutable, and contain behavior.

class Task:
    def __init__(self, id: TaskId, description: str, agent_type: AgentType):
        self._id = id
        self._description = description
        self._agent_type = agent_type
        self._status = TaskStatus.PENDING
        self._started_at: datetime | None = None
        self._completed_at: datetime | None = None

    def start(self) -> None:
        if self._status != TaskStatus.PENDING:
            raise TaskNotPendingException(self._id)
        self._status = TaskStatus.IN_PROGRESS
        self._started_at = datetime.now()

    def complete(self, result: TaskResult) -> None:
        if self._status != TaskStatus.IN_PROGRESS:
            raise TaskNotInProgressException(self._id)
        self._status = TaskStatus.COMPLETED
        self._completed_at = datetime.now()

    @property
    def duration(self) -> timedelta | None:
        if self._started_at and self._completed_at:
            return self._completed_at - self._started_at
        return None