1# domain/entities/user.py
  2from dataclasses import dataclass
  3from datetime import datetime
  4from typing import Optional
  5
  6@dataclass
  7class User:
  8    """Core user entity - no framework dependencies."""
  9    id: str
 10    email: str
 11    name: str
 12    created_at: datetime
 13    is_active: bool = True
 14
 15    def deactivate(self):
 16        """Business rule: deactivating user."""
 17        self.is_active = False
 18
 19    def can_place_order(self) -> bool:
 20        """Business rule: active users can order."""
 21        return self.is_active
 22
 23# domain/interfaces/user_repository.py
 24from abc import ABC, abstractmethod
 25from typing import Optional, List
 26from domain.entities.user import User
 27
 28class IUserRepository(ABC):
 29    """Port: defines contract, no implementation."""
 30
 31    @abstractmethod
 32    async def find_by_id(self, user_id: str) -> Optional[User]:
 33        pass
 34
 35    @abstractmethod
 36    async def find_by_email(self, email: str) -> Optional[User]:
 37        pass
 38
 39    @abstractmethod
 40    async def save(self, user: User) -> User:
 41        pass
 42
 43    @abstractmethod
 44    async def delete(self, user_id: str) -> bool:
 45        pass
 46
 47# use_cases/create_user.py
 48from domain.entities.user import User
 49from domain.interfaces.user_repository import IUserRepository
 50from dataclasses import dataclass
 51from datetime import datetime
 52import uuid
 53
 54@dataclass
 55class CreateUserRequest:
 56    email: str
 57    name: str
 58
 59@dataclass
 60class CreateUserResponse:
 61    user: User
 62    success: bool
 63    error: Optional[str] = None
 64
 65class CreateUserUseCase:
 66    """Use case: orchestrates business logic."""
 67
 68    def __init__(self, user_repository: IUserRepository):
 69        self.user_repository = user_repository
 70
 71    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
 72        # Business validation
 73        existing = await self.user_repository.find_by_email(request.email)
 74        if existing:
 75            return CreateUserResponse(
 76                user=None,
 77                success=False,
 78                error="Email already exists"
 79            )
 80
 81        # Create entity
 82        user = User(
 83            id=str(uuid.uuid4()),
 84            email=request.email,
 85            name=request.name,
 86            created_at=datetime.now(),
 87            is_active=True
 88        )
 89
 90        # Persist
 91        saved_user = await self.user_repository.save(user)
 92
 93        return CreateUserResponse(
 94            user=saved_user,
 95            success=True
 96        )
 97
 98# adapters/repositories/postgres_user_repository.py
 99from domain.interfaces.user_repository import IUserRepository
100from domain.entities.user import User
101from typing import Optional
102import asyncpg
103
104class PostgresUserRepository(IUserRepository):
105    """Adapter: PostgreSQL implementation."""
106
107    def __init__(self, pool: asyncpg.Pool):
108        self.pool = pool
109
110    async def find_by_id(self, user_id: str) -> Optional[User]:
111        async with self.pool.acquire() as conn:
112            row = await conn.fetchrow(
113                "SELECT * FROM users WHERE id = $1", user_id
114            )
115            return self._to_entity(row) if row else None
116
117    async def find_by_email(self, email: str) -> Optional[User]:
118        async with self.pool.acquire() as conn:
119            row = await conn.fetchrow(
120                "SELECT * FROM users WHERE email = $1", email
121            )
122            return self._to_entity(row) if row else None
123
124    async def save(self, user: User) -> User:
125        async with self.pool.acquire() as conn:
126            await conn.execute(
127                """
128                INSERT INTO users (id, email, name, created_at, is_active)
129                VALUES ($1, $2, $3, $4, $5)
130                ON CONFLICT (id) DO UPDATE
131                SET email = $2, name = $3, is_active = $5
132                """,
133                user.id, user.email, user.name, user.created_at, user.is_active
134            )
135            return user
136
137    async def delete(self, user_id: str) -> bool:
138        async with self.pool.acquire() as conn:
139            result = await conn.execute(
140                "DELETE FROM users WHERE id = $1", user_id
141            )
142            return result == "DELETE 1"
143
144    def _to_entity(self, row) -> User:
145        """Map database row to entity."""
146        return User(
147            id=row["id"],
148            email=row["email"],
149            name=row["name"],
150            created_at=row["created_at"],
151            is_active=row["is_active"]
152        )
153
154# adapters/controllers/user_controller.py
155from fastapi import APIRouter, Depends, HTTPException
156from use_cases.create_user import CreateUserUseCase, CreateUserRequest
157from pydantic import BaseModel
158
159router = APIRouter()
160
161class CreateUserDTO(BaseModel):
162    email: str
163    name: str
164
165@router.post("/users")
166async def create_user(
167    dto: CreateUserDTO,
168    use_case: CreateUserUseCase = Depends(get_create_user_use_case)
169):
170    """Controller: handles HTTP concerns only."""
171    request = CreateUserRequest(email=dto.email, name=dto.name)
172    response = await use_case.execute(request)
173
174    if not response.success:
175        raise HTTPException(status_code=400, detail=response.error)
176
177    return {"user": response.user}

 1# Core domain (hexagon center)
 2class OrderService:
 3    """Domain service - no infrastructure dependencies."""
 4
 5    def __init__(
 6        self,
 7        order_repository: OrderRepositoryPort,
 8        payment_gateway: PaymentGatewayPort,
 9        notification_service: NotificationPort
10    ):
11        self.orders = order_repository
12        self.payments = payment_gateway
13        self.notifications = notification_service
14
15    async def place_order(self, order: Order) -> OrderResult:
16        # Business logic
17        if not order.is_valid():
18            return OrderResult(success=False, error="Invalid order")
19
20        # Use ports (interfaces)
21        payment = await self.payments.charge(
22            amount=order.total,
23            customer=order.customer_id
24        )
25
26        if not payment.success:
27            return OrderResult(success=False, error="Payment failed")
28
29        order.mark_as_paid()
30        saved_order = await self.orders.save(order)
31
32        await self.notifications.send(
33            to=order.customer_email,
34            subject="Order confirmed",
35            body=f"Order {order.id} confirmed"
36        )
37
38        return OrderResult(success=True, order=saved_order)
39
40# Ports (interfaces)
41class OrderRepositoryPort(ABC):
42    @abstractmethod
43    async def save(self, order: Order) -> Order:
44        pass
45
46class PaymentGatewayPort(ABC):
47    @abstractmethod
48    async def charge(self, amount: Money, customer: str) -> PaymentResult:
49        pass
50
51class NotificationPort(ABC):
52    @abstractmethod
53    async def send(self, to: str, subject: str, body: str):
54        pass
55
56# Adapters (implementations)
57class StripePaymentAdapter(PaymentGatewayPort):
58    """Primary adapter: connects to Stripe API."""
59
60    def __init__(self, api_key: str):
61        self.stripe = stripe
62        self.stripe.api_key = api_key
63
64    async def charge(self, amount: Money, customer: str) -> PaymentResult:
65        try:
66            charge = self.stripe.Charge.create(
67                amount=amount.cents,
68                currency=amount.currency,
69                customer=customer
70            )
71            return PaymentResult(success=True, transaction_id=charge.id)
72        except stripe.error.CardError as e:
73            return PaymentResult(success=False, error=str(e))
74
75class MockPaymentAdapter(PaymentGatewayPort):
76    """Test adapter: no external dependencies."""
77
78    async def charge(self, amount: Money, customer: str) -> PaymentResult:
79        return PaymentResult(success=True, transaction_id="mock-123")

 1# Value Objects (immutable)
 2from dataclasses import dataclass
 3from typing import Optional
 4
 5@dataclass(frozen=True)
 6class Email:
 7    """Value object: validated email."""
 8    value: str
 9
10    def __post_init__(self):
11        if "@" not in self.value:
12            raise ValueError("Invalid email")
13
14@dataclass(frozen=True)
15class Money:
16    """Value object: amount with currency."""
17    amount: int  # cents
18    currency: str
19
20    def add(self, other: "Money") -> "Money":
21        if self.currency != other.currency:
22            raise ValueError("Currency mismatch")
23        return Money(self.amount + other.amount, self.currency)
24
25# Entities (with identity)
26class Order:
27    """Entity: has identity, mutable state."""
28
29    def __init__(self, id: str, customer: Customer):
30        self.id = id
31        self.customer = customer
32        self.items: List[OrderItem] = []
33        self.status = OrderStatus.PENDING
34        self._events: List[DomainEvent] = []
35
36    def add_item(self, product: Product, quantity: int):
37        """Business logic in entity."""
38        item = OrderItem(product, quantity)
39        self.items.append(item)
40        self._events.append(ItemAddedEvent(self.id, item))
41
42    def total(self) -> Money:
43        """Calculated property."""
44        return sum(item.subtotal() for item in self.items)
45
46    def submit(self):
47        """State transition with business rules."""
48        if not self.items:
49            raise ValueError("Cannot submit empty order")
50        if self.status != OrderStatus.PENDING:
51            raise ValueError("Order already submitted")
52
53        self.status = OrderStatus.SUBMITTED
54        self._events.append(OrderSubmittedEvent(self.id))
55
56# Aggregates (consistency boundary)
57class Customer:
58    """Aggregate root: controls access to entities."""
59
60    def __init__(self, id: str, email: Email):
61        self.id = id
62        self.email = email
63        self._addresses: List[Address] = []
64        self._orders: List[str] = []  # Order IDs, not full objects
65
66    def add_address(self, address: Address):
67        """Aggregate enforces invariants."""
68        if len(self._addresses) >= 5:
69            raise ValueError("Maximum 5 addresses allowed")
70        self._addresses.append(address)
71
72    @property
73    def primary_address(self) -> Optional[Address]:
74        return next((a for a in self._addresses if a.is_primary), None)
75
76# Domain Events
77@dataclass
78class OrderSubmittedEvent:
79    order_id: str
80    occurred_at: datetime = field(default_factory=datetime.now)
81
82# Repository (aggregate persistence)
83class OrderRepository:
84    """Repository: persist/retrieve aggregates."""
85
86    async def find_by_id(self, order_id: str) -> Optional[Order]:
87        """Reconstitute aggregate from storage."""
88        pass
89
90    async def save(self, order: Order):
91        """Persist aggregate and publish events."""
92        await self._persist(order)
93        await self._publish_events(order._events)
94        order._events.clear()