1# architecture-patterns — detailed patterns and worked examples
2 
3## Clean Architecture — Directory Structure
4 
5```
6app/
7├── domain/           # Entities, value objects, interfaces
8│   ├── entities/
9│   │   ├── user.py
10│   │   └── order.py
11│   ├── value_objects/
12│   │   ├── email.py
13│   │   └── money.py
14│   └── interfaces/   # Abstract ports (no implementations)
15│       ├── user_repository.py
16│       └── payment_gateway.py
17├── use_cases/        # Application business rules
18│   ├── create_user.py
19│   ├── process_order.py
20│   └── send_notification.py
21├── adapters/         # Concrete implementations
22│   ├── repositories/
23│   │   ├── postgres_user_repository.py
24│   │   └── redis_cache_repository.py
25│   ├── controllers/
26│   │   └── user_controller.py
27│   └── gateways/
28│       ├── stripe_payment_gateway.py
29│       └── sendgrid_email_gateway.py
30└── infrastructure/   # Framework wiring, config, DI container
31    ├── database.py
32    ├── config.py
33    └── logging.py
34```
35 
36**Dependency rule in one sentence:** every `import` statement in `domain/` and `use_cases/` must point only toward `domain/`; nothing in those layers may import from `adapters/` or `infrastructure/`.
37 
38## Clean Architecture — Core Implementation
39 
40```python
41# domain/entities/user.py
42from dataclasses import dataclass
43from datetime import datetime
44 
45@dataclass
46class User:
47    """Core user entity — no framework dependencies."""
48    id: str
49    email: str
50    name: str
51    created_at: datetime
52    is_active: bool = True
53 
54    def deactivate(self):
55        self.is_active = False
56 
57    def can_place_order(self) -> bool:
58        return self.is_active
59 
60 
61# domain/interfaces/user_repository.py
62from abc import ABC, abstractmethod
63from typing import Optional
64from domain.entities.user import User
65 
66class IUserRepository(ABC):
67    """Port: defines contract, no implementation details."""
68 
69    @abstractmethod
70    async def find_by_id(self, user_id: str) -> Optional[User]: ...
71 
72    @abstractmethod
73    async def find_by_email(self, email: str) -> Optional[User]: ...
74 
75    @abstractmethod
76    async def save(self, user: User) -> User: ...
77 
78    @abstractmethod
79    async def delete(self, user_id: str) -> bool: ...
80 
81 
82# use_cases/create_user.py
83from dataclasses import dataclass
84from datetime import datetime
85from typing import Optional
86import uuid
87from domain.entities.user import User
88from domain.interfaces.user_repository import IUserRepository
89 
90@dataclass
91class CreateUserRequest:
92    email: str
93    name: str
94 
95@dataclass
96class CreateUserResponse:
97    user: Optional[User]
98    success: bool
99    error: Optional[str] = None
100 
101class CreateUserUseCase:
102    """Use case: orchestrates business logic, no HTTP or DB details."""
103 
104    def __init__(self, user_repository: IUserRepository):
105        self.user_repository = user_repository
106 
107    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
108        existing = await self.user_repository.find_by_email(request.email)
109        if existing:
110            return CreateUserResponse(user=None, success=False, error="Email already exists")
111 
112        user = User(
113            id=str(uuid.uuid4()),
114            email=request.email,
115            name=request.name,
116            created_at=datetime.now(),
117        )
118        saved_user = await self.user_repository.save(user)
119        return CreateUserResponse(user=saved_user, success=True)
120 
121 
122# adapters/repositories/postgres_user_repository.py
123from domain.interfaces.user_repository import IUserRepository
124from domain.entities.user import User
125from typing import Optional
126import asyncpg
127 
128class PostgresUserRepository(IUserRepository):
129    """Adapter: PostgreSQL implementation of the user port."""
130 
131    def __init__(self, pool: asyncpg.Pool):
132        self.pool = pool
133 
134    async def find_by_id(self, user_id: str) -> Optional[User]:
135        async with self.pool.acquire() as conn:
136            row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
137            return self._to_entity(row) if row else None
138 
139    async def find_by_email(self, email: str) -> Optional[User]:
140        async with self.pool.acquire() as conn:
141            row = await conn.fetchrow("SELECT * FROM users WHERE email = $1", email)
142            return self._to_entity(row) if row else None
143 
144    async def save(self, user: User) -> User:
145        async with self.pool.acquire() as conn:
146            await conn.execute(
147                """
148                INSERT INTO users (id, email, name, created_at, is_active)
149                VALUES ($1, $2, $3, $4, $5)
150                ON CONFLICT (id) DO UPDATE
151                SET email = $2, name = $3, is_active = $5
152                """,
153                user.id, user.email, user.name, user.created_at, user.is_active,
154            )
155        return user
156 
157    async def delete(self, user_id: str) -> bool:
158        async with self.pool.acquire() as conn:
159            result = await conn.execute("DELETE FROM users WHERE id = $1", user_id)
160            return result == "DELETE 1"
161 
162    def _to_entity(self, row) -> User:
163        return User(
164            id=row["id"], email=row["email"], name=row["name"],
165            created_at=row["created_at"], is_active=row["is_active"],
166        )
167 
168 
169# adapters/controllers/user_controller.py
170from fastapi import APIRouter, Depends, HTTPException
171from pydantic import BaseModel
172from use_cases.create_user import CreateUserUseCase, CreateUserRequest
173 
174router = APIRouter()
175 
176class CreateUserDTO(BaseModel):
177    email: str
178    name: str
179 
180@router.post("/users")
181async def create_user(
182    dto: CreateUserDTO,
183    use_case: CreateUserUseCase = Depends(get_create_user_use_case),
184):
185    """Controller handles HTTP only — no business logic lives here."""
186    response = await use_case.execute(CreateUserRequest(email=dto.email, name=dto.name))
187    if not response.success:
188        raise HTTPException(status_code=400, detail=response.error)
189    return {"user": response.user}
190```
191 
192## Hexagonal Architecture — Ports and Adapters
193 
194```python
195# Core domain service — no infrastructure dependencies
196class OrderService:
197    def __init__(
198        self,
199        order_repository: OrderRepositoryPort,
200        payment_gateway: PaymentGatewayPort,
201        notification_service: NotificationPort,
202    ):
203        self.orders = order_repository
204        self.payments = payment_gateway
205        self.notifications = notification_service
206 
207    async def place_order(self, order: Order) -> OrderResult:
208        if not order.is_valid():
209            return OrderResult(success=False, error="Invalid order")
210 
211        payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
212        if not payment.success:
213            return OrderResult(success=False, error="Payment failed")
214 
215        order.mark_as_paid()
216        saved_order = await self.orders.save(order)
217        await self.notifications.send(
218            to=order.customer_email,
219            subject="Order confirmed",
220            body=f"Order {order.id} confirmed",
221        )
222        return OrderResult(success=True, order=saved_order)
223 
224 
225# Ports (driving and driven interfaces)
226class OrderRepositoryPort(ABC):
227    @abstractmethod
228    async def save(self, order: Order) -> Order: ...
229 
230class PaymentGatewayPort(ABC):
231    @abstractmethod
232    async def charge(self, amount: Money, customer: str) -> PaymentResult: ...
233 
234class NotificationPort(ABC):
235    @abstractmethod
236    async def send(self, to: str, subject: str, body: str): ...
237 
238 
239# Production adapter: Stripe
240class StripePaymentAdapter(PaymentGatewayPort):
241    def __init__(self, api_key: str):
242        import stripe
243        stripe.api_key = api_key
244        self._stripe = stripe
245 
246    async def charge(self, amount: Money, customer: str) -> PaymentResult:
247        try:
248            charge = self._stripe.Charge.create(
249                amount=amount.cents, currency=amount.currency, customer=customer
250            )
251            return PaymentResult(success=True, transaction_id=charge.id)
252        except self._stripe.error.CardError as e:
253            return PaymentResult(success=False, error=str(e))
254 
255 
256# Test adapter: no external dependencies
257class MockPaymentAdapter(PaymentGatewayPort):
258    async def charge(self, amount: Money, customer: str) -> PaymentResult:
259        return PaymentResult(success=True, transaction_id="mock-txn-123")
260```
261 
262## DDD — Value Objects and Aggregates
263 
264```python
265# Value Objects: immutable, validated at construction
266from dataclasses import dataclass
267 
268@dataclass(frozen=True)
269class Email:
270    value: str
271 
272    def __post_init__(self):
273        if "@" not in self.value or "." not in self.value.split("@")[-1]:
274            raise ValueError(f"Invalid email: {self.value}")
275 
276@dataclass(frozen=True)
277class Money:
278    amount: int   # cents
279    currency: str
280 
281    def __post_init__(self):
282        if self.amount < 0:
283            raise ValueError("Money amount cannot be negative")
284        if self.currency not in {"USD", "EUR", "GBP"}:
285            raise ValueError(f"Unsupported currency: {self.currency}")
286 
287    def add(self, other: "Money") -> "Money":
288        if self.currency != other.currency:
289            raise ValueError("Currency mismatch")
290        return Money(self.amount + other.amount, self.currency)
291 
292 
293# Aggregate root: enforces all invariants for its cluster of entities
294class Order:
295    def __init__(self, id: str, customer_id: str):
296        self.id = id
297        self.customer_id = customer_id
298        self.items: list[OrderItem] = []
299        self.status = OrderStatus.PENDING
300        self._events: list[DomainEvent] = []
301 
302    def add_item(self, product: Product, quantity: int):
303        if self.status != OrderStatus.PENDING:
304            raise ValueError("Cannot modify a submitted order")
305        item = OrderItem(product=product, quantity=quantity)
306        self.items.append(item)
307        self._events.append(ItemAddedEvent(order_id=self.id, item=item))
308 
309    @property
310    def total(self) -> Money:
311        totals = [item.subtotal() for item in self.items]
312        return sum(totals[1:], totals[0]) if totals else Money(0, "USD")
313 
314    def submit(self):
315        if not self.items:
316            raise ValueError("Cannot submit an empty order")
317        if self.status != OrderStatus.PENDING:
318            raise ValueError("Order already submitted")
319        self.status = OrderStatus.SUBMITTED
320        self._events.append(OrderSubmittedEvent(order_id=self.id))
321 
322    def pop_events(self) -> list[DomainEvent]:
323        events, self._events = self._events, []
324        return events
325 
326 
327# Repository: persist and reconstitute aggregates
328class OrderRepository(ABC):
329    @abstractmethod
330    async def find_by_id(self, order_id: str) -> Optional[Order]: ...
331 
332    @abstractmethod
333    async def save(self, order: Order) -> None: ...
334    # Implementations persist events via pop_events() after writing state
335```
336