1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# ❌ IMPLEMENTATION-DRIVEN: Class mirrors database table
 
class Order:
    """Directly mirrors 'orders' table. No domain logic."""
    
    def __init__(self):
        # Default constructor for ORM hydration
        self.id: int = 0
        self.customer_id: int = 0
        self.status: int = 0           # Magic numbers!
        self.total_cents: int = 0
        self.shipping_address_line1: str = ""
        self.shipping_address_line2: str = ""
        self.shipping_city: str = ""
        self.shipping_zip: str = ""
        self.created_at: datetime = None
        self.updated_at: datetime = None
    
    # Just getters and setters — anemic model
    def get_status(self) -> int:
        return self.status
    
    def set_status(self, status: int) -> None:
        # No validation, no business rules
        self.status = status
    
    # ... more getters and setters
 
# Business logic forced into a "service"
class OrderService:
    """Logic that should be in Order is here instead."""
    
    def ship_order(self, order: Order) -> None:
        if order.status != 1:  # What does 1 mean?!
            raise Exception("Cannot ship")
        
        # Validation scattered here, not in domain
        if not order.shipping_address_line1:
            raise Exception("No address")
        
        order.set_status(2)  # And what's 2?
        # Hope nobody forgets to call this service and just sets status directly

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

# ✅ DOMAIN-DRIVEN: Class models the business concept
 
from enum import Enum
from dataclasses import dataclass
from typing import List
 
class OrderStatus(Enum):
    """Self-documenting states with business meaning."""
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"
 
 
@dataclass(frozen=True)
class Address:
    """Value Object: Grouped, validated, immutable."""
    line1: str
    line2: str
    city: str
    postal_code: str
    country: str
    
    def __post_init__(self):
        if not self.line1 or not self.city or not self.postal_code:
            raise ValueError("Address requires line1, city, and postal_code")
 
 
@dataclass(frozen=True)
class Money:
    """Value Object: Prevents floating point errors, handles currency."""
    amount_cents: int
    currency: str = "USD"
    
    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return Money(self.amount_cents + other.amount_cents, self.currency)
 
 
class Order:
    """
    Rich domain model: Encapsulates data AND behavior.
    Models the business concept, not the database.
    """
    
    def __init__(
        self, 
        order_id: str,
        customer_id: str,
        shipping_address: Address,
        items: List["OrderItem"]
    ):
        if not items:
            raise ValueError("Order must have at least one item")
        
        self._id = order_id
        self._customer_id = customer_id
        self._shipping_address = shipping_address
        self._items = list(items)  # Defensive copy
        self._status = OrderStatus.PENDING
    
    @property
    def order_id(self) -> str:
        return self._id
    
    @property
    def status(self) -> OrderStatus:
        return self._status
    
    @property
    def total(self) -> Money:
        """Calculate total from items — derived, not stored."""
        total = Money(0)
        for item in self._items:
            total = total.add(item.subtotal)
        return total
    
    def confirm(self) -> None:
        """
        Customer confirms the order. 
        Business rule: Can only confirm pending orders.
        """
        if self._status != OrderStatus.PENDING:
            raise OrderException(
                f"Cannot confirm order in {self._status.value} status"
            )
        self._status = OrderStatus.CONFIRMED
    
    def ship(self, tracking_number: str) -> None:
        """
        Order is shipped.
        Business rule: Can only ship confirmed orders.
        """
        if self._status != OrderStatus.CONFIRMED:
            raise OrderException(
                f"Cannot ship order in {self._status.value} status"
            )
        if not self._shipping_address:
            raise OrderException("Cannot ship without address")
        
        self._status = OrderStatus.SHIPPED
        self._tracking_number = tracking_number
    
    def cancel(self, reason: str) -> None:
        """
        Cancel the order.
        Business rule: Cannot cancel shipped or delivered orders.
        """
        if self._status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
            raise OrderException(
                f"Cannot cancel {self._status.value} order"
            )
        self._status = OrderStatus.CANCELLED
        self._cancellation_reason = reason
 
 
class OrderException(Exception):
    """Domain-specific exception for order rule violations."""
    pass
 
 
@dataclass
class OrderItem:
    product_id: str
    quantity: int
    unit_price: Money
    
    @property
    def subtotal(self) -> Money:
        return Money(self.unit_price.amount_cents * self.quantity)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

# Domain Layer: Pure business objects, no infrastructure
 
class Order:
    """Pure domain entity. Knows nothing about persistence."""
    
    def __init__(self, order_id: str, customer_id: str, items: list):
        self._id = order_id
        self._customer_id = customer_id
        self._items = items
        self._status = OrderStatus.PENDING
    
    def confirm(self) -> None:
        # Pure business logic
        if self._status != OrderStatus.PENDING:
            raise OrderException("Cannot confirm")
        self._status = OrderStatus.CONFIRMED
 
 
# Domain Layer: Repository interface (abstract, no implementation)
from abc import ABC, abstractmethod
 
class OrderRepository(ABC):
    """
    Abstract interface in domain layer.
    Defines WHAT the domain needs, not HOW it's implemented.
    """
    
    @abstractmethod
    def find_by_id(self, order_id: str) -> Order | None:
        """Retrieve order by ID. Returns None if not found."""
        pass
    
    @abstractmethod
    def save(self, order: Order) -> None:
        """Persist order changes."""
        pass
    
    @abstractmethod
    def find_pending_orders(self) -> list[Order]:
        """Find all orders awaiting processing."""
        pass
 
 
# Infrastructure Layer: Concrete implementation
 
class PostgresOrderRepository(OrderRepository):
    """
    Infrastructure implementation of the domain interface.
    This class knows about databases; the domain does not.
    """
    
    def __init__(self, connection):
        self._conn = connection
    
    def find_by_id(self, order_id: str) -> Order | None:
        # SQL is here, not in domain
        row = self._conn.execute(
            "SELECT * FROM orders WHERE id = %s", 
            (order_id,)
        ).fetchone()
        
        if not row:
            return None
        
        # Translate from database row to domain object
        return self._hydrate_order(row)
    
    def save(self, order: Order) -> None:
        # Translate from domain object to database row
        self._conn.execute(
            """
            INSERT INTO orders (id, customer_id, status, ...) 
            VALUES (%s, %s, %s, ...)
            ON CONFLICT (id) DO UPDATE SET status = %s, ...
            """,
            (
                order.order_id,
                order._customer_id,
                order.status.value,  # Store enum as string
                ...
            )
        )
    
    def _hydrate_order(self, row) -> Order:
        """Convert database row to domain object."""
        # Handle the translation here, including:
        # - Converting status string to enum
        # - Loading related items
        # - Converting address columns to Address object
        
        items = self._load_items(row["id"])
        address = Address(
            line1=row["shipping_line1"],
            line2=row["shipping_line2"],
            city=row["shipping_city"],
            postal_code=row["shipping_zip"],
            country=row["shipping_country"]
        )
        
        order = Order.__new__(Order)  # Bypass __init__ for hydration
        order._id = row["id"]
        order._customer_id = row["customer_id"]
        order._status = OrderStatus(row["status"])
        order._shipping_address = address
        order._items = items
        return order
 
 
# Application Layer: Use case that coordinates domain and infrastructure
 
class ConfirmOrderUseCase:
    """
    Application service orchestrating the use case.
    Coordinates between domain and infrastructure.
    """
    
    def __init__(self, order_repo: OrderRepository, email_service):
        self._order_repo = order_repo
        self._email_service = email_service
    
    def execute(self, order_id: str) -> None:
        # Load domain object via repository
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise NotFoundError(f"Order {order_id} not found")
        
        # Execute domain logic
        order.confirm()  # Pure business logic in the domain
        
        # Persist changes via repository
        self._order_repo.save(order)
        
        # Side effects via infrastructure services
        self._email_service.send_confirmation(order)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

# Domain Model: Pure, no framework pollution
 
class Customer:
    """Domain entity with encapsulated state and behavior."""
    
    def __init__(self, customer_id: str, email: EmailAddress, name: str):
        self._id = customer_id
        self._email = email  # Value Object
        self._name = name
        self._status = CustomerStatus.ACTIVE
    
    def suspend(self, reason: str) -> None:
        if self._status == CustomerStatus.SUSPENDED:
            raise CustomerException("Already suspended")
        self._status = CustomerStatus.SUSPENDED
        self._suspension_reason = reason
    
    @property
    def email(self) -> EmailAddress:
        return self._email
 
 
# Persistence Model: Framework-specific, annotation-heavy
 
from sqlalchemy import Column, String, Enum as SQLEnum
from sqlalchemy.ext.declarative import declarative_base
 
Base = declarative_base()
 
class CustomerRecord(Base):
    """
    Persistence model for SQLAlchemy.
    This class is NOT used in business logic—only for storage.
    """
    __tablename__ = "customers"
    
    id = Column(String, primary_key=True)
    email = Column(String, nullable=False)  # Stored as string
    name = Column(String, nullable=False)
    status = Column(String, nullable=False)
    suspension_reason = Column(String, nullable=True)
    
    # Framework requirements satisfied here
    def __init__(self):
        pass  # Empty constructor for ORM
 
 
# Repository translates between domain and persistence
 
class SQLAlchemyCustomerRepository(CustomerRepository):
    def find_by_id(self, customer_id: str) -> Customer | None:
        record = self._session.query(CustomerRecord).get(customer_id)
        if not record:
            return None
        return self._to_domain(record)
    
    def save(self, customer: Customer) -> None:
        record = self._to_record(customer)
        self._session.merge(record)
        self._session.flush()
    
    def _to_domain(self, record: CustomerRecord) -> Customer:
        """Convert persistence model to domain model."""
        customer = Customer.__new__(Customer)
        customer._id = record.id
        customer._email = EmailAddress(record.email)  # Wrap in Value Object
        customer._name = record.name
        customer._status = CustomerStatus(record.status)
        customer._suspension_reason = record.suspension_reason
        return customer
    
    def _to_record(self, customer: Customer) -> CustomerRecord:
        """Convert domain model to persistence model."""
        record = CustomerRecord()
        record.id = customer._id
        record.email = customer._email.value  # Unwrap Value Object
        record.name = customer._name
        record.status = customer._status.value
        record.suspension_reason = getattr(customer, '_suspension_reason', None)
        return record