1---
2name: architecture-patterns
3description: Implement proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design. Use this skill when designing clean architecture for a new microservice, when refactoring a monolith to use bounded contexts, when implementing hexagonal or onion architecture patterns, or when debugging dependency cycles between application layers.
4---
5 
6# Architecture Patterns
7 
8Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems.
9 
10**Given:** a service boundary or module to architect.
11**Produces:** layered structure with clear dependency rules, interface definitions, and test boundaries.
12 
13## When to Use This Skill
14 
15- Designing new backend services or microservices from scratch
16- Refactoring monolithic applications where business logic is entangled with ORM models or HTTP concerns
17- Establishing bounded contexts before splitting a system into services
18- Debugging dependency cycles where infrastructure code bleeds into the domain layer
19- Creating testable codebases where use-case tests do not require a running database
20- Implementing domain-driven design tactical patterns (aggregates, value objects, domain events)
21 
22## Core Concepts
23 
24### 1. Clean Architecture (Uncle Bob)
25 
26**Layers (dependency flows inward):**
27 
28- **Entities**: Core business models, no framework imports
29- **Use Cases**: Application business rules, orchestrate entities
30- **Interface Adapters**: Controllers, presenters, gateways — translate between use cases and external formats
31- **Frameworks & Drivers**: UI, database, external services — all at the outermost ring
32 
33**Key Principles:**
34 
35- Dependencies point inward only; inner layers know nothing about outer layers
36- Business logic is independent of frameworks, databases, and delivery mechanisms
37- Every layer boundary is crossed via an abstract interface
38- Testable without UI, database, or external services
39 
40### 2. Hexagonal Architecture (Ports and Adapters)
41 
42**Components:**
43 
44- **Domain Core**: Business logic lives here, framework-free
45- **Ports**: Abstract interfaces that define how the core interacts with the outside world (driving and driven)
46- **Adapters**: Concrete implementations of ports (PostgreSQL adapter, Stripe adapter, REST adapter)
47 
48**Benefits:**
49 
50- Swap implementations without touching the core (e.g., replace PostgreSQL with DynamoDB)
51- Use in-memory adapters in tests — no Docker required
52- Technology decisions deferred to the edges
53 
54### 3. Domain-Driven Design (DDD)
55 
56**Strategic Patterns:**
57 
58- **Bounded Contexts**: Isolate a coherent model for one subdomain; avoid sharing a single model across the whole system
59- **Context Mapping**: Define how contexts relate (Anti-Corruption Layer, Shared Kernel, Open Host Service)
60- **Ubiquitous Language**: Every term in code matches the term used by domain experts
61 
62**Tactical Patterns:**
63 
64- **Entities**: Objects with stable identity that change over time
65- **Value Objects**: Immutable objects identified by their attributes (Email, Money, Address)
66- **Aggregates**: Consistency boundaries; only the root is accessible from outside
67- **Repositories**: Persist and reconstitute aggregates; abstract over the storage mechanism
68- **Domain Events**: Capture things that happened inside the domain; used for cross-aggregate coordination
69 
70## Clean Architecture — Directory Structure
71 
72```
73app/
74├── domain/           # Entities, value objects, interfaces
75│   ├── entities/
76│   │   ├── user.py
77│   │   └── order.py
78│   ├── value_objects/
79│   │   ├── email.py
80│   │   └── money.py
81│   └── interfaces/   # Abstract ports (no implementations)
82│       ├── user_repository.py
83│       └── payment_gateway.py
84├── use_cases/        # Application business rules
85│   ├── create_user.py
86│   ├── process_order.py
87│   └── send_notification.py
88├── adapters/         # Concrete implementations
89│   ├── repositories/
90│   │   ├── postgres_user_repository.py
91│   │   └── redis_cache_repository.py
92│   ├── controllers/
93│   │   └── user_controller.py
94│   └── gateways/
95│       ├── stripe_payment_gateway.py
96│       └── sendgrid_email_gateway.py
97└── infrastructure/   # Framework wiring, config, DI container
98    ├── database.py
99    ├── config.py
100    └── logging.py
101```
102 
103**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/`.
104 
105## Clean Architecture — Core Implementation
106 
107```python
108# domain/entities/user.py
109from dataclasses import dataclass
110from datetime import datetime
111 
112@dataclass
113class User:
114    """Core user entity — no framework dependencies."""
115    id: str
116    email: str
117    name: str
118    created_at: datetime
119    is_active: bool = True
120 
121    def deactivate(self):
122        self.is_active = False
123 
124    def can_place_order(self) -> bool:
125        return self.is_active
126 
127 
128# domain/interfaces/user_repository.py
129from abc import ABC, abstractmethod
130from typing import Optional
131from domain.entities.user import User
132 
133class IUserRepository(ABC):
134    """Port: defines contract, no implementation details."""
135 
136    @abstractmethod
137    async def find_by_id(self, user_id: str) -> Optional[User]: ...
138 
139    @abstractmethod
140    async def find_by_email(self, email: str) -> Optional[User]: ...
141 
142    @abstractmethod
143    async def save(self, user: User) -> User: ...
144 
145    @abstractmethod
146    async def delete(self, user_id: str) -> bool: ...
147 
148 
149# use_cases/create_user.py
150from dataclasses import dataclass
151from datetime import datetime
152from typing import Optional
153import uuid
154from domain.entities.user import User
155from domain.interfaces.user_repository import IUserRepository
156 
157@dataclass
158class CreateUserRequest:
159    email: str
160    name: str
161 
162@dataclass
163class CreateUserResponse:
164    user: Optional[User]
165    success: bool
166    error: Optional[str] = None
167 
168class CreateUserUseCase:
169    """Use case: orchestrates business logic, no HTTP or DB details."""
170 
171    def __init__(self, user_repository: IUserRepository):
172        self.user_repository = user_repository
173 
174    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
175        existing = await self.user_repository.find_by_email(request.email)
176        if existing:
177            return CreateUserResponse(user=None, success=False, error="Email already exists")
178 
179        user = User(
180            id=str(uuid.uuid4()),
181            email=request.email,
182            name=request.name,
183            created_at=datetime.now(),
184        )
185        saved_user = await self.user_repository.save(user)
186        return CreateUserResponse(user=saved_user, success=True)
187 
188 
189# adapters/repositories/postgres_user_repository.py
190from domain.interfaces.user_repository import IUserRepository
191from domain.entities.user import User
192from typing import Optional
193import asyncpg
194 
195class PostgresUserRepository(IUserRepository):
196    """Adapter: PostgreSQL implementation of the user port."""
197 
198    def __init__(self, pool: asyncpg.Pool):
199        self.pool = pool
200 
201    async def find_by_id(self, user_id: str) -> Optional[User]:
202        async with self.pool.acquire() as conn:
203            row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
204            return self._to_entity(row) if row else None
205 
206    async def find_by_email(self, email: str) -> Optional[User]:
207        async with self.pool.acquire() as conn:
208            row = await conn.fetchrow("SELECT * FROM users WHERE email = $1", email)
209            return self._to_entity(row) if row else None
210 
211    async def save(self, user: User) -> User:
212        async with self.pool.acquire() as conn:
213            await conn.execute(
214                """
215                INSERT INTO users (id, email, name, created_at, is_active)
216                VALUES ($1, $2, $3, $4, $5)
217                ON CONFLICT (id) DO UPDATE
218                SET email = $2, name = $3, is_active = $5
219                """,
220                user.id, user.email, user.name, user.created_at, user.is_active,
221            )
222        return user
223 
224    async def delete(self, user_id: str) -> bool:
225        async with self.pool.acquire() as conn:
226            result = await conn.execute("DELETE FROM users WHERE id = $1", user_id)
227            return result == "DELETE 1"
228 
229    def _to_entity(self, row) -> User:
230        return User(
231            id=row["id"], email=row["email"], name=row["name"],
232            created_at=row["created_at"], is_active=row["is_active"],
233        )
234 
235 
236# adapters/controllers/user_controller.py
237from fastapi import APIRouter, Depends, HTTPException
238from pydantic import BaseModel
239from use_cases.create_user import CreateUserUseCase, CreateUserRequest
240 
241router = APIRouter()
242 
243class CreateUserDTO(BaseModel):
244    email: str
245    name: str
246 
247@router.post("/users")
248async def create_user(
249    dto: CreateUserDTO,
250    use_case: CreateUserUseCase = Depends(get_create_user_use_case),
251):
252    """Controller handles HTTP only — no business logic lives here."""
253    response = await use_case.execute(CreateUserRequest(email=dto.email, name=dto.name))
254    if not response.success:
255        raise HTTPException(status_code=400, detail=response.error)
256    return {"user": response.user}
257```
258 
259## Hexagonal Architecture — Ports and Adapters
260 
261```python
262# Core domain service — no infrastructure dependencies
263class OrderService:
264    def __init__(
265        self,
266        order_repository: OrderRepositoryPort,
267        payment_gateway: PaymentGatewayPort,
268        notification_service: NotificationPort,
269    ):
270        self.orders = order_repository
271        self.payments = payment_gateway
272        self.notifications = notification_service
273 
274    async def place_order(self, order: Order) -> OrderResult:
275        if not order.is_valid():
276            return OrderResult(success=False, error="Invalid order")
277 
278        payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
279        if not payment.success:
280            return OrderResult(success=False, error="Payment failed")
281 
282        order.mark_as_paid()
283        saved_order = await self.orders.save(order)
284        await self.notifications.send(
285            to=order.customer_email,
286            subject="Order confirmed",
287            body=f"Order {order.id} confirmed",
288        )
289        return OrderResult(success=True, order=saved_order)
290 
291 
292# Ports (driving and driven interfaces)
293class OrderRepositoryPort(ABC):
294    @abstractmethod
295    async def save(self, order: Order) -> Order: ...
296 
297class PaymentGatewayPort(ABC):
298    @abstractmethod
299    async def charge(self, amount: Money, customer: str) -> PaymentResult: ...
300 
301class NotificationPort(ABC):
302    @abstractmethod
303    async def send(self, to: str, subject: str, body: str): ...
304 
305 
306# Production adapter: Stripe
307class StripePaymentAdapter(PaymentGatewayPort):
308    def __init__(self, api_key: str):
309        import stripe
310        stripe.api_key = api_key
311        self._stripe = stripe
312 
313    async def charge(self, amount: Money, customer: str) -> PaymentResult:
314        try:
315            charge = self._stripe.Charge.create(
316                amount=amount.cents, currency=amount.currency, customer=customer
317            )
318            return PaymentResult(success=True, transaction_id=charge.id)
319        except self._stripe.error.CardError as e:
320            return PaymentResult(success=False, error=str(e))
321 
322 
323# Test adapter: no external dependencies
324class MockPaymentAdapter(PaymentGatewayPort):
325    async def charge(self, amount: Money, customer: str) -> PaymentResult:
326        return PaymentResult(success=True, transaction_id="mock-txn-123")
327```
328 
329## DDD — Value Objects and Aggregates
330 
331```python
332# Value Objects: immutable, validated at construction
333from dataclasses import dataclass
334 
335@dataclass(frozen=True)
336class Email:
337    value: str
338 
339    def __post_init__(self):
340        if "@" not in self.value or "." not in self.value.split("@")[-1]:
341            raise ValueError(f"Invalid email: {self.value}")
342 
343@dataclass(frozen=True)
344class Money:
345    amount: int   # cents
346    currency: str
347 
348    def __post_init__(self):
349        if self.amount < 0:
350            raise ValueError("Money amount cannot be negative")
351        if self.currency not in {"USD", "EUR", "GBP"}:
352            raise ValueError(f"Unsupported currency: {self.currency}")
353 
354    def add(self, other: "Money") -> "Money":
355        if self.currency != other.currency:
356            raise ValueError("Currency mismatch")
357        return Money(self.amount + other.amount, self.currency)
358 
359 
360# Aggregate root: enforces all invariants for its cluster of entities
361class Order:
362    def __init__(self, id: str, customer_id: str):
363        self.id = id
364        self.customer_id = customer_id
365        self.items: list[OrderItem] = []
366        self.status = OrderStatus.PENDING
367        self._events: list[DomainEvent] = []
368 
369    def add_item(self, product: Product, quantity: int):
370        if self.status != OrderStatus.PENDING:
371            raise ValueError("Cannot modify a submitted order")
372        item = OrderItem(product=product, quantity=quantity)
373        self.items.append(item)
374        self._events.append(ItemAddedEvent(order_id=self.id, item=item))
375 
376    @property
377    def total(self) -> Money:
378        totals = [item.subtotal() for item in self.items]
379        return sum(totals[1:], totals[0]) if totals else Money(0, "USD")
380 
381    def submit(self):
382        if not self.items:
383            raise ValueError("Cannot submit an empty order")
384        if self.status != OrderStatus.PENDING:
385            raise ValueError("Order already submitted")
386        self.status = OrderStatus.SUBMITTED
387        self._events.append(OrderSubmittedEvent(order_id=self.id))
388 
389    def pop_events(self) -> list[DomainEvent]:
390        events, self._events = self._events, []
391        return events
392 
393 
394# Repository: persist and reconstitute aggregates
395class OrderRepository(ABC):
396    @abstractmethod
397    async def find_by_id(self, order_id: str) -> Optional[Order]: ...
398 
399    @abstractmethod
400    async def save(self, order: Order) -> None: ...
401    # Implementations persist events via pop_events() after writing state
402```
403 
404## Testing — In-Memory Adapters
405 
406The hallmark of correctly applied Clean Architecture is that every use case can be exercised in a plain unit test with no real database, no Docker, and no network:
407 
408```python
409# tests/unit/test_create_user.py
410import asyncio
411from typing import Dict, Optional
412from domain.entities.user import User
413from domain.interfaces.user_repository import IUserRepository
414from use_cases.create_user import CreateUserUseCase, CreateUserRequest
415 
416 
417class InMemoryUserRepository(IUserRepository):
418    def __init__(self):
419        self._store: Dict[str, User] = {}
420 
421    async def find_by_id(self, user_id: str) -> Optional[User]:
422        return self._store.get(user_id)
423 
424    async def find_by_email(self, email: str) -> Optional[User]:
425        return next((u for u in self._store.values() if u.email == email), None)
426 
427    async def save(self, user: User) -> User:
428        self._store[user.id] = user
429        return user
430 
431    async def delete(self, user_id: str) -> bool:
432        return self._store.pop(user_id, None) is not None
433 
434 
435async def test_create_user_succeeds():
436    repo = InMemoryUserRepository()
437    use_case = CreateUserUseCase(user_repository=repo)
438 
439    response = await use_case.execute(CreateUserRequest(email="[email protected]", name="Alice"))
440 
441    assert response.success
442    assert response.user.email == "[email protected]"
443    assert response.user.id is not None
444 
445 
446async def test_duplicate_email_rejected():
447    repo = InMemoryUserRepository()
448    use_case = CreateUserUseCase(user_repository=repo)
449 
450    await use_case.execute(CreateUserRequest(email="[email protected]", name="Alice"))
451    response = await use_case.execute(CreateUserRequest(email="[email protected]", name="Alice2"))
452 
453    assert not response.success
454    assert "already exists" in response.error
455```
456 
457## Troubleshooting
458 
459### Use case tests require a running database
460 
461Business logic has leaked into the infrastructure layer. Move all database calls behind an `IRepository` interface and inject an in-memory implementation in tests (see Testing section above). The use case constructor must accept the abstract port, not the concrete class.
462 
463### Circular imports between layers
464 
465A common symptom is `ImportError: cannot import name X` between `use_cases` and `adapters`. This happens when a use case imports a concrete adapter class instead of the abstract port. Enforce the rule: `use_cases/` imports only from `domain/` (entities and interfaces). It must never import from `adapters/` or `infrastructure/`.
466 
467### Framework decorators appearing in domain entities
468 
469If SQLAlchemy `Column()` or Pydantic `Field()` annotations appear on domain entities, the entity is no longer pure. Create a separate ORM model in `adapters/repositories/` and map to/from the domain entity in the repository's `_to_entity()` method.
470 
471### All logic ending up in controllers
472 
473When the controller grows beyond HTTP parsing and response formatting, extract the logic into a use case class. A controller method should do three things only: parse the request, call a use case, map the response.
474 
475### Value objects raising errors too late
476 
477Validate invariants in `__post_init__` (Python) or the constructor so an invalid `Email` or `Money` cannot be constructed at all. This surfaces bad data at the boundary, not deep inside business logic.
478 
479### Context bleed across bounded contexts
480 
481If the `Order` context is importing `User` entities from the `Identity` context, introduce an Anti-Corruption Layer. The `Order` context should hold its own lightweight `CustomerId` value object and only call the `Identity` context through an explicit interface.
482 
483## Advanced Patterns
484 
485For detailed DDD bounded context mapping, full multi-service project trees, Anti-Corruption Layer implementations, and Onion Architecture comparisons, see:
486 
487- [`references/advanced-patterns.md`](references/advanced-patterns.md)
488 
489## Related Skills
490 
491- `microservices-patterns` — Apply these architecture patterns when decomposing a monolith into services
492- `cqrs-implementation` — Use Clean Architecture as the structural foundation for CQRS command/query separation
493- `saga-orchestration` — Sagas require well-defined aggregate boundaries, which DDD tactical patterns provide
494- `event-store-design` — Domain events produced by aggregates feed directly into an event store
495