Architecture Patterns

---

name: architecture-patterns model: reasoning

---

Architecture Patterns

WHAT

Backend architecture patterns for building maintainable, testable systems: Clean Architecture, Hexagonal Architecture, and Domain-Driven Design.

WHEN

Designing new backend systems from scratch

Refactoring monoliths for better maintainability

Establishing architecture standards for teams

Creating testable, mockable codebases

Planning microservices decomposition

KEYWORDS

clean architecture, hexagonal, ports and adapters, DDD, domain-driven design, layers, entities, use cases, repositories, aggregates, bounded contexts

---

Decision Framework: Which Pattern?

| Situation | Recommended Pattern | |-----------|---------------------| | Simple CRUD app | None (over-engineering) | | Medium complexity, team standardization | Clean Architecture | | Multiple external integrations that change frequently | Hexagonal (Ports & Adapters) | | Complex business domain with many rules | Domain-Driven Design | | Large system with multiple teams | DDD + Bounded Contexts |

Quick Reference

Clean Architecture Layers

┌──────────────────────────────────────┐
│      Frameworks & Drivers (UI, DB)   │  ← Outer: Can change
├──────────────────────────────────────┤
│      Interface Adapters              │  ← Controllers, Gateways
├──────────────────────────────────────┤
│      Use Cases                       │  ← Application Logic
├──────────────────────────────────────┤
│      Entities                        │  ← Core Business Rules
└──────────────────────────────────────┘

Dependency Rule: Dependencies point INWARD only. Inner layers never import outer layers.

Hexagonal Architecture

         ┌─────────────┐
    ┌────│   Adapter   │────┐    (REST API)
    │    └─────────────┘    │
    ▼                       ▼
┌──────┐              ┌──────────┐
│ Port │◄────────────►│  Domain  │
└──────┘              └──────────┘
    ▲                       ▲
    │    ┌─────────────┐    │
    └────│   Adapter   │────┘    (Database)
         └─────────────┘

Ports: Interfaces defining what the domain needs Adapters: Implementations (swappable for testing)

---

Directory Structure

app/
├── domain/           # Entities & business rules (innermost)
│   ├── entities/
│   │   └── user.py
│   ├── value_objects/
│   │   └── email.py
│   └── interfaces/   # Ports
│       └── user_repository.py
├── use_cases/        # Application business rules
│   └── create_user.py
├── adapters/         # Interface implementations
│   ├── repositories/
│   │   └── postgres_user_repository.py
│   └── controllers/
│       └── user_controller.py
└── infrastructure/   # Framework & external concerns
    ├── database.py
    └── config.py

---

Pattern 1: Clean Architecture

Entity (Domain Layer)

from dataclasses import dataclass
from datetime import datetime
@dataclass
class User:
    """Core entity - NO framework dependencies."""
    id: str
    email: str
    name: str
    created_at: datetime
    is_active: bool = True
    def deactivate(self):
        """Business rule in entity."""
        self.is_active = False
    def can_place_order(self) -> bool:
        return self.is_active

Port (Interface)

from abc import ABC, abstractmethod
from typing import Optional
class IUserRepository(ABC):
    """Port: defines contract, no implementation."""
    
    @abstractmethod
    async def find_by_id(self, user_id: str) -> Optional[User]:
        pass
    
    @abstractmethod
    async def save(self, user: User) -> User:
        pass

Use Case (Application Layer)

@dataclass
class CreateUserRequest:
    email: str
    name: str
@dataclass  
class CreateUserResponse:
    user: Optional[User]
    success: bool
    error: Optional[str] = None
class CreateUserUseCase:
    """Use case: orchestrates business logic."""
    
    def __init__(self, user_repository: IUserRepository):
        self.user_repository = user_repository  # Injected dependency
    
    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
        # Business validation
        existing = await self.user_repository.find_by_email(request.email)
        if existing:
            return CreateUserResponse(user=None, success=False, error="Email exists")
        
        # Create entity
        user = User(
            id=str(uuid.uuid4()),
            email=request.email,
            name=request.name,
            created_at=datetime.now()
        )
        
        saved = await self.user_repository.save(user)
        return CreateUserResponse(user=saved, success=True)

Adapter (Implementation)

class PostgresUserRepository(IUserRepository):
    """Adapter: PostgreSQL implementation of the port."""
    
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
    
    async def find_by_id(self, user_id: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT * FROM users WHERE id = $1", user_id
            )
            return self._to_entity(row) if row else None
    
    async def save(self, user: User) -> User:
        async with self.pool.acquire() as conn:
            await conn.execute(
                """INSERT INTO users (id, email, name, created_at, is_active)
                   VALUES ($1, $2, $3, $4, $5)
                   ON CONFLICT (id) DO UPDATE SET email=$2, name=$3, is_active=$5""",
                user.id, user.email, user.name, user.created_at, user.is_active
            )
            return user

---

Pattern 2: Hexagonal (Ports & Adapters)

Best when you have multiple external integrations that may change.

# Domain Service (Core)
class OrderService:
    def __init__(
        self,
        order_repo: OrderRepositoryPort,      # Port
        payment: PaymentGatewayPort,          # Port
        notifications: NotificationPort       # Port
    ):
        self.orders = order_repo
        self.payments = payment
        self.notifications = notifications
    
    async def place_order(self, order: Order) -> OrderResult:
        # Pure business logic - no infrastructure details
        if not order.is_valid():
            return OrderResult(success=False, error="Invalid order")
        
        payment = await self.payments.charge(order.total, order.customer_id)
        if not payment.success:
            return OrderResult(success=False, error="Payment failed")
        
        order.mark_as_paid()
        saved = await self.orders.save(order)
        await self.notifications.send(order.customer_email, "Order confirmed")
        
        return OrderResult(success=True, order=saved)
Adapters (swap these for testing or changing providers)
class StripePaymentAdapter(PaymentGatewayPort):
    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        # Real Stripe implementation
        ...
class MockPaymentAdapter(PaymentGatewayPort):
    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        return PaymentResult(success=True, transaction_id="mock-123")

---

Pattern 3: Domain-Driven Design

For complex business domains with many rules.

Value Objects (Immutable)

@dataclass(frozen=True)
class Email:
    """Value object: validated, immutable."""
    value: str
    
    def __post_init__(self):
        if "@" not in self.value:
            raise ValueError("Invalid email")
@dataclass(frozen=True)
class Money:
    amount: int  # cents
    currency: str
    
    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Currency mismatch")
        return Money(self.amount + other.amount, self.currency)

Aggregates (Consistency Boundaries)

class Order:
    """Aggregate root: enforces invariants."""
    
    def __init__(self, id: str, customer: Customer):
        self.id = id
        self.customer = customer
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING
        self._events: List[DomainEvent] = []
    
    def add_item(self, product: Product, quantity: int):
        """Business logic in aggregate."""
        if quantity > product.max_quantity:
            raise ValueError(f"Max {product.max_quantity} allowed")
        
        item = OrderItem(product, quantity)
        self.items.append(item)
        self._events.append(ItemAddedEvent(self.id, item))
    
    def submit(self):
        """State transition with invariant enforcement."""
        if not self.items:
            raise ValueError("Cannot submit empty order")
        if self.status != OrderStatus.PENDING:
            raise ValueError("Order already submitted")
        
        self.status = OrderStatus.SUBMITTED
        self._events.append(OrderSubmittedEvent(self.id))

Repository Pattern

class OrderRepository:
    """Persist/retrieve aggregates, publish domain events."""
    
    async def save(self, order: Order):
        await self._persist(order)
        await self._publish_events(order._events)
        order._events.clear()

---

Testing Benefits

All patterns enable the same testing approach:

# Test with mock adapter
async def test_create_user():
    mock_repo = MockUserRepository()
    use_case = CreateUserUseCase(user_repository=mock_repo)
    
    result = await use_case.execute(CreateUserRequest(
        email="test@example.com",
        name="Test User"
    ))
    
    assert result.success
    assert result.user.email == "test@example.com"

---

NEVER

Anemic Domain Models: Entities with only data, no behavior (put logic IN entities)

Framework Coupling: Business logic importing Flask, FastAPI, Django ORM

Fat Controllers: Business logic in HTTP handlers

Leaky Abstractions: Repository returning ORM objects instead of domain entities

Skipping Layers: Controller directly accessing database

Over-Engineering: Using Clean Architecture for simple CRUD apps

Circular Dependencies: Use cases importing controllers