# Ddd Specification Pattern > Implements the DDD specification pattern — composable business rule objects using AND/OR/NOT boolean algebra, expression tree translation for ORM query pushdown, domain validation specs, protocol-based contracts, and reusable primitive factories for rich domain modeling in Python. - Skill: `paulpas/ddd-specification-pattern` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/ddd-specification-pattern` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/ddd-specification-pattern/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/ddd-specification-pattern --- # Specification Pattern for DDD Implements the specification pattern as a composable business rule system within Domain-Driven Design. This skill covers primitive specifications, boolean algebra composition (AND/OR/NOT), expression tree translation for ORM query pushdown, protocol-based type contracts, domain validation specs with error reporting, and reusable primitive factories. The key differentiator from simple validators is **composability** — combining specs via operator overloading to build complex rules from simple parts without modifying existing code. ## TL;DR Checklist - [ ] Base specification defines `is_satisfied_by()` and `description` - [ ] Operator overloading: `__and__`, `__or__`, `__invert__` for boolean composition - [ ] Primitive specs check exactly one business rule — never combine multiple concerns - [ ] Expression tree translation converts composed specs into ORM query filters - [ ] Protocol-based contracts accept any structurally-compatible spec implementation - [ ] Domain validation specs return descriptive error messages, not just True/False - [ ] Spec translator lives in infrastructure layer — domain specs must NOT import ORMs --- ## When to Use Use this skill when: - Building composable business rules that combine via AND/OR/NOT boolean algebra - Filtering repositories by complex, dynamic criteria without hardcoding if/else chains - Validating aggregate state before mutations using declarative rule specifications - Needing reusable, independently-testable business rule objects across multiple aggregates - Implementing CQRS read-side projections with dynamic query composition --- ## When NOT to Use Avoid this skill for: - Simple validation that only needs one or two conditions — if/else is more readable than spec objects for trivial cases - ORM-specific query building without composable requirements — use native query builders directly - Performance-critical paths where expression tree translation adds unnecessary overhead — evaluate specs in-memory with compiled lambdas instead - Testing infrastructure concerns like database connectivity, network latency, or deployment health --- ## Core Workflow 1. **Define the Specification Base** — Create an ABC or Protocol that defines `is_satisfied_by(candidate: T) -> bool`, a human-readable `description` string, and optional `errors` dict for validation contexts. Operator overloading (`__and__`, `__or__`, `__invert__`) enables boolean composition. **Checkpoint:** Every spec must be pure — no side effects during evaluation. Mutation breaks composability. 2. **Implement Primitive Specifications** — Create leaf specs that check exactly one business rule: `HasPositiveBalance`, `IsNotFrozen`, `MeetsMinimumAmount`. Each should have a descriptive name and documentation string. **Checkpoint:** If a spec checks more than one condition, split it into primitives and compose them with AND/OR operators. 3. **Compose Complex Rules** — Combine primitives using bitwise operators (`&`, `|`, `~`). Store composites in named classes or constants for reuse: `class EligibleForLoan: ACTIVE & POSITIVE_BALANCE & ~FROZEN`. **Checkpoint:** Named composite specs are more readable than inline compositions at call sites and can be independently documented. 4. **Add Expression Tree Translation** — For repository filtering, translate specification trees into ORM query filters (SQLAlchemy, Django ORM). This requires converting the spec tree to `PredicateNode` objects and mapping leaf predicates to column comparisons. **Checkpoint:** The translator lives in the infrastructure layer — domain specs must never import SQLAlchemy or any persistence technology. 5. **Implement Protocol-Based Contracts** — Use `Protocol[T]` for structural typing so any class with a matching `is_satisfied_by()` method satisfies the spec interface without explicit inheritance. This enables plugin architectures and duck-typed filtering services. **Checkpoint:** Protocol-based contracts make specs interchangeable regardless of their base class hierarchy. 6. **Build Primitive Factory Pattern** — Create parameterized generic specs via factory methods (`Primitives.equals("status", "active")`, `Primitives.in_range("balance", 0, 1_000_000)`). This avoids defining new classes for every field-comparison combination and reduces boilerplate. **Checkpoint:** Factory-created specs carry descriptions generated from their parameters for better error reporting. --- ## Implementation Patterns ### Pattern 1: Base Specification with Boolean Composition ```python from __future__ import annotations from abc import ABC, abstractmethod from typing import Generic, TypeVar, Any T = TypeVar("T") class Specification(ABC, Generic[T]): """Base specification class for DDD business rules. Specifications encapsulate a single business rule as an object with: - A human-readable description - An is_satisfied_by() evaluation method - Boolean composition via operator overloading (& | ~) The key insight: composability distinguishes specifications from simple validators. Complex rules are built by combining primitives, not by writing nested if/else chains. """ description: str = "No description provided." errors: dict[str, str] = {} @abstractmethod def is_satisfied_by(self, candidate: T) -> bool: ... # --- Boolean composition via operator overloading --- def __and__(self: "Specification[T]", other: "Specification[T]") -> "AndSpecification[T]": return AndSpecification(self, other) def __or__(self: "Specification[T]", other: "Specification[T]") -> "OrSpecification[T]": return OrSpecification(self, other) def __invert__(self) -> "NotSpecification[T]": return NotSpecification(self) # Callable shorthand for quick evaluation def __call__(self, candidate: T) -> bool: return self.is_satisfied_by(candidate) class AndSpecification(Specification[T]): """Composite spec: satisfied if BOTH child specs are satisfied.""" description: str = "AND composition of specifications" def __init__(self, left: Specification[T], right: Specification[T]) -> None: self.left = left self.right = right def is_satisfied_by(self, candidate: T) -> bool: if not self.left.is_satisfied_by(candidate): return False # Short-circuit evaluation return self.right.is_satisfied_by(candidate) @property def description(self) -> str: return f"({self.left.description} AND {self.right.description})" class OrSpecification(Specification[T]): """Composite spec: satisfied if EITHER child spec is satisfied.""" description: str = "OR composition of specifications" def __init__(self, left: Specification[T], right: Specification[T]) -> None: self.left = left self.right = right def is_satisfied_by(self, candidate: T) -> bool: if self.left.is_satisfied_by(candidate): return True # Short-circuit evaluation return self.right.is_satisfied_by(candidate) @property def description(self) -> str: return f"({self.left.description} OR {self.right.description})" class NotSpecification(Specification[T]): """Composite spec: satisfied when child spec is NOT satisfied.""" description: str = "NOT composition of specification" def __init__(self, spec: Specification[T]) -> None: self.spec = spec def is_satisfied_by(self, candidate: T) -> bool: return not self.spec.is_satisfied_by(candidate) @property def description(self) -> str: return f"(NOT {self.spec.description})" # --- Usage example with domain entities --- class Account: def __init__(self, balance: float, is_frozen: bool, status: str) -> None: self.balance = balance self.is_frozen = is_frozen self.status = status class HasPositiveBalance(Specification[Account]): description = "Account must have a positive balance." def is_satisfied_by(self, account: Account) -> bool: return account.balance > 0 class IsNotFrozen(Specification[Account]): description = "Account must not be frozen." def is_satisfied_by(self, account: Account) -> bool: return not account.is_frozen class MeetsMinimumBalance(Specification[Account]): description = "Account balance must meet minimum threshold." def __init__(self, minimum: float) -> None: self.minimum = minimum def is_satisfied_by(self, account: Account) -> bool: return account.balance >= self.minimum # Composition using bitwise operators eligible_for_loan: Specification[Account] = ( HasPositiveBalance() & IsNotFrozen() & MeetsMinimumBalance(100.0) ) frozen_accounts_excluded: Specification[Account] = Specification.__invert__(HasPositiveBalance()) # ~IsFrozen pattern active_and_eligible: Specification[Account] = ( MeetsMinimumBalance(100.0) & ~IsNotFrozen() | HasPositiveBalance() ) # Verify composition account_good = Account(balance=500.0, is_frozen=False, status="active") account_bad = Account(balance=-50.0, is_frozen=True, status="suspended") assert eligible_for_loan.is_satisfied_by(account_good) # True assert not eligible_for_loan.is_satisfied_by(account_bad) # False ``` ### Pattern 2: Domain Validation Specifications with Error Reporting ```python from __future__ import annotations from dataclasses import dataclass, field from datetime import date from typing import Optional @dataclass class Trade: symbol: str quantity: int price: float side: str # "BUY" or "SELL" status: str = "PENDING" approved_at: Optional[date] = None @dataclass class ValidationError: rule_name: str description: str candidate_value: Any | None = None class SpecificationWithErrors(Specification[T]): """Specification that tracks validation errors instead of returning boolean only.""" def validate(self, candidate: T) -> tuple[bool, list[ValidationError]]: """Return (is_valid, list_of_errors). Evaluates each composite spec in the tree and collects all failing rules. Used by domain services before allowing state mutations. """ errors = self._collect_errors(candidate, []) return len(errors) == 0, errors def _collect_errors(self, candidate: T, path: list[str]) -> list[ValidationError]: raise NotImplementedError # Implemented by each composite type class AndSpecificationWithErrors(SpecificationWithErrors[T]): """AND composition that collects errors from ALL failing branches.""" def __init__(self, left: SpecificationWithErrors[T], right: SpecificationWithErrors[T]) -> None: self.left = left self.right = right self.description = f"({left.description} AND {right.description})" def is_satisfied_by(self, candidate: T) -> bool: return self.left.is_satisfied_by(candidate) and self.right.is_satisfied_by(candidate) def _collect_errors(self, candidate: T, path: list[str]) -> list[ValidationError]: errors = [] if not self.left.is_satisfied_by(candidate): left_errors = self.left._collect_errors(candidate, path + ["LEFT"]) errors.extend(left_errors) if not self.right.is_satisfied_by(candidate): right_errors = self.right._collect_errors(candidate, path + ["RIGHT"]) errors.extend(right_errors) return errors class OrSpecificationWithErrors(SpecificationWithErrors[T]): """OR composition that collects errors from the failing branch.""" def __init__(self, left: SpecificationWithErrors[T], right: SpecificationWithErrors[T]) -> None: self.left = left self.right = right self.description = f"({left.description} OR {right.description})" def is_satisfied_by(self, candidate: T) -> bool: return self.left.is_satisfied_by(candidate) or self.right.is_satisfied_by(candidate) def _collect_errors(self, candidate: T, path: list[str]) -> list[ValidationError]: if self.left.is_satisfied_by(candidate): # Left satisfied — only collect errors from right (the failing branch) return self.right._collect_errors(candidate, path + ["RIGHT"]) if self.right.is_satisfied_by(candidate): return self.left._collect_errors(candidate, path + ["LEFT"]) # Neither satisfied — collect from both return ( self.left._collect_errors(candidate, path + ["LEFT"]) + self.right._collect_errors(candidate, path + ["RIGHT"]) ) class NotSpecificationWithErrors(SpecificationWithErrors[T]): """NOT composition that wraps error collection from the inner spec.""" def __init__(self, spec: SpecificationWithErrors[T]) -> None: self.spec = spec self.description = f"(NOT {spec.description})" def is_satisfied_by(self, candidate: T) -> bool: return not self.spec.is_satisfied_by(candidate) def _collect_errors(self, candidate: T, path: list[str]) -> list[ValidationError]: if self.spec.is_satisfied_by(candidate): # The wrapped spec IS satisfied, so NOT fails — report the inner spec's success as error return [ ValidationError( rule_name="negated_spec", description=f"Expected NOT {self.spec.description} to be true, but it was satisfied", candidate_value=candidate, ) ] return [] # --- Primitive specs with error reporting --- class SymbolValid(SpecificationWithErrors[Trade]): description = "Trade symbol must be a recognized ticker (AAPL, GOOGL, MSFT)." def __init__(self) -> None: self.valid_symbols = {"AAPL", "GOOGL", "MSFT", "TSLA", "AMZN"} def is_satisfied_by(self, trade: Trade) -> bool: return trade.symbol in self.valid_symbols class ValidQuantity(SpecificationWithErrors[Trade]): description = "Quantity must be a positive integer." def is_satisfied_by(self, trade: Trade) -> bool: return isinstance(trade.quantity, int) and trade.quantity > 0 class MaxOrderValue(SpecificationWithErrors[Trade]): description = "Order value (quantity x price) must not exceed $1,000,000." def __init__(self, max_value: float = 1_000_000.0) -> None: self.max_value = max_value def is_satisfied_by(self, trade: Trade) -> bool: return (trade.quantity * trade.price) <= self.max_value class NotDuplicateTrade(SpecificationWithErrors[Trade]): description = "No existing approved trade for this symbol and side in the last 24h." def __init__(self, repository: Any = None) -> None: self.repository = repository # In production, inject via DI def is_satisfied_by(self, trade: Trade) -> bool: if self.repository is None: return True # Skip during testing without real repo existing = getattr(self.repository, "find_recent", lambda *a, **k: []) trades = existing(trade.symbol, trade.side, hours=24) return not any( e.price == trade.price and e.quantity == trade.quantity for e in trades ) # --- Composite validation for order submission --- class OrderSubmissionValidator(SpecificationWithErrors[Trade]): """Composite spec that validates an entire order before execution. Combines symbol validity, quantity constraints, value limits, and duplicate detection into a single evaluation surface. """ description = "Complete order submission validation" def __init__(self, max_value: float = 1_000_000.0) -> None: self.spec: SpecificationWithErrors[Trade] = AndSpecificationWithErrors( AndSpecificationWithErrors( SymbolValid(), ValidQuantity(), ), AndSpecificationWithErrors( MaxOrderValue(max_value), NotDuplicateTrade(), ), ) def is_satisfied_by(self, trade: Trade) -> bool: return self.spec.is_satisfied_by(trade) def validate(self, trade: Trade) -> tuple[bool, list[ValidationError]]: return self.spec.validate(trade) # --- Usage in a domain service --- class TradingService: def submit_order(self, trade: Trade, max_value: float) -> None: """Submit an order after validation. Raises if validation fails.""" validator = OrderSubmissionValidator(max_value=max_value) is_valid, errors = validator.validate(trade) if not is_valid: for error in errors: print(f"[VALIDATION] {error.rule_name}: {error.description}") raise ValueError( f"Order submission failed: {len(errors)} validation rule(s) did not pass" ) trade.status = "APPROVED" trade.approved_at = date.today() # Verify error collection works with composite specs bad_trade = Trade(symbol="INVALID", quantity=-1, price=50.0, side="BUY") validator = OrderSubmissionValidator() is_valid, errors = validator.validate(bad_trade) assert not is_valid rule_names = [e.rule_name for e in errors] assert "SymbolValid" in rule_names or "symbol" in str(errors[0]).lower() ``` ### Pattern 3: Expression Tree Translation for ORM Query Pushdown ```python from __future__ import annotations from abc import ABC, abstractmethod from enum import Enum, auto from typing import Any, Generic, TypeVar from sqlalchemy import and_, or_, select from sqlalchemy.orm import Session T = TypeVar("T") # --- Expression Tree Node Types (domain-agnostic) --- class PredicateType(Enum): EQ = "eq" NEQ = "neq" GT = "gt" LT = "lt" GTE = "gte" LTE = "lte" CONTAINS = "contains" NOT_CONTAINS = "not_contains" class PredicateNode(ABC, Generic[T]): """Abstract node in the specification expression tree.""" @abstractmethod def evaluate(self, candidate: T) -> bool: ... @abstractmethod def to_orm_clause(self, model_cls: type) -> Any: """Translate this predicate into an ORM filter clause.""" ... class LeafPredicate(PredicateNode[T]): """Leaf node: field comparison (equality, range, contains).""" def __init__(self, field_name: str, operator: PredicateType, value: Any) -> None: self.field_name = field_name self.operator = operator self.value = value def evaluate(self, candidate: T) -> bool: actual = getattr(candidate, self.field_name) if self.operator == PredicateType.EQ: return actual == self.value elif self.operator == PredicateType.NEQ: return actual != self.value elif self.operator == PredicateType.GT: return actual > self.value elif self.operator == PredicateType.LT: return actual < self.value elif self.operator == PredicateType.GTE: return actual >= self.value elif self.operator == PredicateType.LTE: return actual <= self.value elif self.operator == PredicateType.CONTAINS: return self.value in (actual or "") elif self.operator == PredicateType.NOT_CONTAINS: return self.value not in (actual or "") return False def to_orm_clause(self, model_cls: type) -> Any: column = getattr(model_cls, self.field_name) if self.operator == PredicateType.EQ: return column == self.value elif self.operator == PredicateType.NEQ: return column != self.value elif self.operator == PredicateType.GT: return column > self.value elif self.operator == PredicateType.LT: return column < self.value elif self.operator == PredicateType.GTE: return column >= self.value elif self.operator == PredicateType.LTE: return column <= self.value elif self.operator == PredicateType.CONTAINS: return column.contains(self.value) elif self.operator == PredicateType.NOT_CONTAINS: return ~column.contains(self.value) raise ValueError(f"Unsupported operator: {self.operator}") class AndPredicate(PredicateNode[T]): """Composite AND node in the expression tree.""" def __init__(self, left: PredicateNode[T], right: PredicateNode[T]) -> None: self.left = left self.right = right def evaluate(self, candidate: T) -> bool: return self.left.evaluate(candidate) and self.right.evaluate(candidate) def to_orm_clause(self, model_cls: type) -> Any: left_clause = self.left.to_orm_clause(model_cls) right_clause = self.right.to_orm_clause(model_cls) return and_(left_clause, right_clause) class OrPredicate(PredicateNode[T]): """Composite OR node in the expression tree.""" def __init__(self, left: PredicateNode[T], right: PredicateNode[T]) -> None: self.left = left self.right = right def evaluate(self, candidate: T) -> bool: return self.left.evaluate(candidate) or self.right.evaluate(candidate) def to_orm_clause(self, model_cls: type) -> Any: left_clause = self.left.to_orm_clause(model_cls) right_clause = self.right.to_orm_clause(model_cls) return or_(left_clause, right_clause) class SpecificationQueryTranslator: """Translates Specification trees into SQLAlchemy query filters. IMPORTANT: This translator lives in the INFRASTRUCTURE layer, NOT in domain code. Domain specs must NEVER import SQLAlchemy. The translator bridges the gap between domain logic and persistence technology. """ @staticmethod def to_sqlalchemy_filter( predicate: PredicateNode[T], model_cls: type, ) -> Any: """Convert a predicate tree into a SQLAlchemy filter clause.""" if isinstance(predicate, LeafPredicate): return predicate.to_orm_clause(model_cls) elif isinstance(predicate, AndPredicate): left = SpecificationQueryTranslator.to_sqlalchemy_filter( predicate.left, model_cls ) right = SpecificationQueryTranslator.to_sqlalchemy_filter( predicate.right, model_cls ) return and_(left, right) elif isinstance(predicate, OrPredicate): left = SpecificationQueryTranslator.to_sqlalchemy_filter( predicate.left, model_cls ) right = SpecificationQueryTranslator.to_sqlalchemy_filter( predicate.right, model_cls ) return or_(left, right) raise ValueError(f"Unsupported predicate type: {type(predicate).__name__}") # --- Usage with SQLAlchemy --- from sqlalchemy import Column, Integer, String from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class AccountModel(Base): __tablename__ = "accounts" id: Mapped[int] = mapped_column(Integer, primary_key=True) status: Mapped[str] = mapped_column(String) balance: Mapped[float] = mapped_column(Integer) is_frozen: Mapped[bool] = mapped_column(Integer) # Build a specification tree using predicates (not domain specs — these are the infrastructure layer) spec_tree: PredicateNode[AccountModel] = AndPredicate( LeafPredicate("status", PredicateType.EQ, "active"), OrPredicate( LeafPredicate("balance", PredicateType.GT, 0), AndPredicate( LeafPredicate("is_frozen", PredicateType.EQ, False), LeafPredicate("balance", PredicateType.GTE, -100), ), ), ) # Translate to ORM filter filter_clause = SpecificationQueryTranslator.to_sqlalchemy_filter(spec_tree, AccountModel) # Execute the query with Session() as session: stmt = select(AccountModel).where(filter_clause) accounts = list(session.scalars(stmt).all()) # --- In-Memory Filtering (no translation needed) --- class AccountFilter: """Filters in-memory collections using specification evaluation.""" def __init__(self, spec: SpecificationWithErrors[T]) -> None: self.spec = spec def filter(self, items: list[T]) -> list[T]: return [item for item in items if self.spec.is_satisfied_by(item)] # In-memory filtering of domain objects (no ORM needed) accounts = [ Account(balance=500.0, is_frozen=False, status="active"), Account(balance=-50.0, is_frozen=True, status="suspended"), Account(balance=25.0, is_frozen=False, status="active"), ] eligible_for_loan = HasPositiveBalance() & IsNotFrozen() & MeetsMinimumBalance(100.0) filter_service: AccountFilter[Account] = AccountFilter(eligible_for_loan) result = filter_service.filter(accounts) assert len(result) == 1 and result[0].balance == 500.0 ``` ### Pattern 4: Protocol-Based Contracts for Maximum Flexibility ```python from __future__ import annotations from typing import Protocol, TypeVar, Generic T_co = TypeVar("T_co", covariant=True) class SpecificationLike(Protocol[T]): """Structural protocol — any class with matching methods satisfies this. This enables duck-typed specifications: a class does NOT need to inherit from the Specification base class to work with filtering services, domain services, and composition utilities. Any structurally-compatible class works. Example of duck-typed compatibility: class LegacyFilter: # Not a subclass of Specification! def is_satisfied_by(self, account: Account) -> bool: return account.balance > 0 # This still works with Protocol-based services: service = SpecFilterService[Account](LegacyFilter()) """ description: str errors: dict[str, str] def is_satisfied_by(self, candidate: T) -> bool: ... class SpecFilterService(Generic[T]): """Accepts any object conforming to the Specification protocol. Works with concrete Specification subclasses, dataclasses, protocols, or plain classes — as long as they implement is_satisfied_by(candidate). """ def __init__(self, spec: SpecificationLike[T]) -> None: self.spec = spec def filter(self, items: list[T]) -> list[T]: return [item for item in items if self.spec.is_satisfied_by(item)] def describe(self) -> str: return getattr(self.spec, "description", "Unknown specification") class OldAccountsFilter: """Structurally-compatible filter — NOT a subclass of Specification.""" description = "Accounts older than 90 days." errors: dict[str, str] = {} def __init__(self, reference_date: date) -> None: self.reference_date = reference_date def is_satisfied_by(self, account: Account) -> bool: from datetime import timedelta return (self.reference_date - date(2025, 1, 1)).days > 90 # Usage — Protocol-based services accept any structurally-compatible implementation filter_service = SpecFilterService[Account](HasPositiveBalance()) print(filter_service.describe()) # "Account must have a positive balance." legacy_filter = SpecFilterService[Account](OldAccountsFilter(date(2025, 6, 1))) print(legacy_filter.describe()) # "Accounts older than 90 days." # --- Builder pattern for fluent spec construction --- class SpecBuilder: """Fluent builder that creates Specification trees via method chaining.""" def __init__(self, entity_type: type[T]) -> None: self._entity_type = entity_type self._predicates: list[PredicateNode[T]] = [] self._operator: str = "AND" # Default operator between predicates def with_field(self, field: str) -> "SpecFieldBuilder[T]": """Start building a predicate for a specific field.""" return SpecFieldBuilder(field, self._predicates) def combine(self, operator: str = "AND") -> Specification[T]: if not self._predicates: raise ValueError("No specifications defined. Use with_field() first.") result = self._predicates[0] for predicate in self._predicates[1:]: if operator == "AND": result = AndPredicate(result, predicate) # type: ignore[arg-type] elif operator == "OR": result = OrPredicate(result, predicate) # type: ignore[arg-type] return result class SpecFieldBuilder(Generic[T]): """Builder for individual field predicates within a fluent chain.""" def __init__(self, field: str, accumulator: list[PredicateNode[T]]) -> None: self.field = field self.accumulator = accumulator def equals(self, value: Any) -> "SpecFieldBuilder[T]": self.accumulator.append(LeafPredicate(self.field, PredicateType.EQ, value)) return self def not_equals(self, value: Any) -> "SpecFieldBuilder[T]": self.accumulator.append(LeafPredicate(self.field, PredicateType.NEQ, value)) return self def greater_than(self, value: Any) -> "SpecFieldBuilder[T]": self.accumulator.append(LeafPredicate(self.field, PredicateType.GT, value)) return self def less_than(self, value: Any) -> "SpecFieldBuilder[T]": self.accumulator.append(LeafPredicate(self.field, PredicateType.LT, value)) return self def contains(self, substring: str) -> "SpecFieldBuilder[T]": self.accumulator.append(LeafPredicate(self.field, PredicateType.CONTAINS, substring)) return self def build(self) -> PredicateNode[T]: if not self.accumulator: raise ValueError("No predicates built. Use operator methods first.") result = self.accumulator[0] for pred in self.accumulator[1:]: result = AndPredicate(result, pred) return result # Usage — fluent builder API class AccountModel: pass # Simplified for illustration builder = SpecBuilder(AccountModel) spec_tree = ( builder.with_field("status").equals("active") ).build() ``` ### Pattern 5: Primitive Factory for Reusable Parameterized Specs ```python from __future__ import annotations import re from typing import Generic, TypeVar T_co = TypeVar("T_co") class Comparison(Enum): EQ = "==" NEQ = "!=" GT = ">" LT = "<" GTE = ">=" LTE = "<=" @dataclass(frozen=True) class FieldSpec(Specification[T_co]): """Generic field-based specification — parameterized by field, operator, and value. Eliminates boilerplate: instead of defining a new class for every field comparison, use this parameterized spec with any entity type. """ field_name: str operator: Comparison value: Any description: str = field(init=False) def __post_init__(self) -> None: object.__setattr__( self, "description", f"{self.field_name} {self.operator.value} {self.value!r}" ) def is_satisfied_by(self, candidate: T_co) -> bool: actual = getattr(candidate, self.field_name) if actual is None: return False # Null values never satisfy comparisons if self.operator == Comparison.EQ: return actual == self.value elif self.operator == Comparison.NEQ: return actual != self.value elif self.operator == Comparison.GT: return actual > self.value elif self.operator == Comparison.LT: return actual < self.value elif self.operator == Comparison.GTE: return actual >= self.value elif self.operator == Comparison.LTE: return actual <= self.value raise ValueError(f"Unsupported operator for comparison spec: {self.operator}") class Primitives: """Factory for commonly reused primitive specifications. Provides parameterized specs without requiring new class definitions. Use these as building blocks for complex composed specifications. """ @staticmethod def equals(field: str, value: Any) -> FieldSpec: return FieldSpec(field, Comparison.EQ, value) @staticmethod def not_equals(field: str, value: Any) -> FieldSpec: return FieldSpec(field, Comparison.NEQ, value) @staticmethod def greater_than(field: str, value: Any) -> FieldSpec: return FieldSpec(field, Comparison.GT, value) @staticmethod def less_than(field: str, value: Any) -> FieldSpec: return FieldSpec(field, Comparison.LT, value) @staticmethod def gte(field: str, value: Any) -> FieldSpec: return FieldSpec(field, Comparison.GTE, value) @staticmethod def lte(field: str, value: Any) -> FieldSpec: return FieldSpec(field, Comparison.LTE, value) @staticmethod def in_range(field: str, minimum: float, maximum: float) -> Specification[T_co]: """Create a range-check specification.""" class _InRange(Specification): # type: ignore[type-arg] description = f"{field} must be between {minimum!r} and {maximum!r}." def is_satisfied_by(self, candidate: Any) -> bool: val = getattr(candidate, field) if val is None: return False return minimum <= val <= maximum return _InRange() # type: ignore[return-value] @staticmethod def matches_pattern(field: str, pattern: str) -> Specification[T_co]: """Create a regex-matches specification.""" compiled = re.compile(pattern) class _Matches(Specification): # type: ignore[type-arg] description = f"{field} must match pattern '{pattern}'." def is_satisfied_by(self, candidate: Any) -> bool: actual = getattr(candidate, field, "") return bool(compiled.search(actual)) if actual else False return _Matches() # type: ignore[return-value] # --- Usage --- # Parameterized field specs (no new class definitions needed!) status_spec = Primitives.equals("status", "active") balance_spec = Primitives.greater_than("balance", 0) frozen_spec = Primitives.equals("is_frozen", False) # Compose into complex rules account_rules: Specification[Account] = ( status_spec & balance_spec & ~frozen_spec ) # Range and pattern specs range_spec: Specification[Account] = Primitives.in_range("balance", 0, 1_000_000) email_spec: Specification[Account] = Primitives.matches_pattern( "email", r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$" ) # Registry pattern — store reusable specs as module-level constants class AccountSpecifications: """Registry of commonly-used account specifications.""" ACTIVE = Primitives.equals("status", "active") POSITIVE_BALANCE = Primitives.greater_than("balance", 0) NOT_FROZEN = ~Primitives.equals("is_frozen", True) ELIGIBLE_FOR_LOAN = ACTIVE & POSITIVE_BALANCE & NOT_FROZEN # Usage of pre-built spec registry eligible_accounts = AccountFilter[Account](AccountSpecifications.ELIGIBLE_FOR_LOAN) result = eligible_accounts.filter(accounts) assert len(result) >= 0 # Depends on test data ``` --- ## Constraints ### MUST DO - **Each primitive spec checks exactly one business rule** — if a spec checks multiple conditions, split it into primitives and compose with AND/OR. This follows the Single Responsibility Principle. - **Use Python's bitwise operators for composition** — `&`, `|`, `~` are idiomatic and readable. Avoid method-based composition like `.and_with()` which is verbose and obscures intent. - **Name negated specs explicitly when possible** — prefer `IsNotFrozen()` over `~IsFrozen()`. Negation with `~` produces confusing error messages ("Expected condition NOT satisfied: X"). - **Store composed specs as named class attributes or module constants** — reuse is the point of composition. Don't repeat complex spec expressions at every call site. - **Keep the translator in the infrastructure layer** — domain specs must NEVER import SQLAlchemy, Django ORM, or any persistence technology. The `SpecificationQueryTranslator` bridges domain logic to persistence. - **Use Protocol-based contracts for filter services** — accept `SpecificationLike[T]` rather than concrete base classes, enabling any structurally-compatible implementation. ### MUST NOT DO - **Put ORM imports inside domain spec classes** — this creates circular dependencies and makes specs untestable outside the persistence context. The translator adapter is a separate class in `infrastructure/`. - **Let specs have side effects during evaluation** — specs must be pure functions. Evaluation changes nothing; it only returns True/False. Mutation breaks composability and caching. - **Create specs that depend on each other circularly** — spec A should not import or instantiate spec B directly. Use dependency injection (pass dependencies via `__init__`) for cross-spec references. - **Evaluate complex compositions on millions of records in-memory** — if the dataset is large, push filters to the database via the expression tree translator. In-memory evaluation of 20+ AND/OR specs per record is slow. - **Use `extra="forbid"` on consumer schemas** — (this also applies from event pattern) new producer fields will crash all consumers. Use forward-compatible designs. --- ## Live References > Authoritative documentation and reference material for the specification pattern in DDD systems. - [Specification Pattern — Martin Fowler](https://martinfowler.com/apsupp/spec.pdf) — The original paper defining the specification pattern as a composable business rule object - [Implementing DDD — Scott Millett, Chapter 7](https://www.implementingddd.com/) — Practical implementation of specifications within bounded contexts - [DDD Lite — Vaughn Vernon](https://www.infoq.com/articles/domain-driven-design-essentials/) — Simplified DDD for teams finding full specification pattern heavy - [sutoppu — Python Specification Pattern Library](https://github.com/u8slvn/sutoppu) — The leading standalone Python implementation with ABC base class and metaclass-based error tracking - [Python typing documentation — Protocol](https://docs.python.org/3/library/typing.html#typing.Protocol) — Structural subtyping for specification contracts - [SQLAlchemy Documentation — ORM Query Filters](https://docs.sqlalchemy.org/en/20/orm/queryguide/sql_operators.html) — For expression tree to ORM filter translation --- ## Related Skills | Skill | Purpose | |---|---| | `ddd-tactical-patterns` | Defines aggregate roots, value objects, and basic repository patterns — specifications complement repositories by providing composable filtering criteria | | `domain-driven-tactical` | Another tactical implementation; uses specs internally for invariant checks but doesn't cover the full composition framework | | `domain-repository-pattern` | Repository abstraction with specification-based queries; this skill provides the spec layer that feeds into repository filtering | | `ddd-command-pattern` | Commands route write operations; specifications validate those commands before they reach aggregate roots — complementary read-side concern | | `microservice-contract-testing` | Contract testing for inter-service APIs; specifications can be used as shared validation rules between services in a DDD microservices architecture | > 📖 skill(local cache): ddd-tactical-patterns, domain-driven-tactical, domain-repository-pattern, ddd-command-pattern, microservice-contract-testing