Architecture Patterns

Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems.

When to Use This Skill

• Designing new backend systems from scratch

• Refactoring monolithic applications for better maintainability

• Establishing architecture standards for your team

• Migrating from tightly coupled to loosely coupled architectures

• Implementing domain-driven design principles

• Creating testable and mockable codebases

• Planning microservices decomposition

Core Concepts

1. Clean Architecture (Uncle Bob)

Layers (dependency flows inward):

• Entities: Core business models

• Use Cases: Application business rules

• Interface Adapters: Controllers, presenters, gateways

• Frameworks & Drivers: UI, database, external services

Key Principles:

• Dependencies point inward

• Inner layers know nothing about outer layers

• Business logic independent of frameworks

• Testable without UI, database, or external services

2. Hexagonal Architecture (Ports and Adapters)

Components:

• Domain Core: Business logic

• Ports: Interfaces defining interactions

• Adapters: Implementations of ports (database, REST, message queue)

Benefits:

• Swap implementations easily (mock for testing)

• Technology-agnostic core

• Clear separation of concerns

3. Domain-Driven Design (DDD)

Strategic Patterns:

• Bounded Contexts: Separate models for different domains

• Context Mapping: How contexts relate

• Ubiquitous Language: Shared terminology

Tactical Patterns:

• Entities: Objects with identity

• Value Objects: Immutable objects defined by attributes

• Aggregates: Consistency boundaries

• Repositories: Data access abstraction

• Domain Events: Things that happened

Clean Architecture Pattern

Directory Structure

app/ ├── domain/ # Entities & business rules │ ├── entities/ │ │ ├── user.py │ │ └── order.py │ ├── value_objects/ │ │ ├── email.py │ │ └── money.py │ └── interfaces/ # Abstract interfaces │ ├── user_repository.py │ └── payment_gateway.py ├── use_cases/ # Application business rules │ ├── create_user.py │ ├── process_order.py │ └── send_notification.py ├── adapters/ # Interface implementations │ ├── repositories/ │ │ ├── postgres_user_repository.py │ │ └── redis_cache_repository.py │ ├── controllers/ │ │ └── user_controller.py │ └── gateways/ │ ├── stripe_payment_gateway.py │ └── sendgrid_email_gateway.py └── infrastructure/ # Framework & external concerns ├── database.py ├── config.py └── logging.py

Implementation Example

domain/entities/user.py

from dataclasses import dataclass from datetime import datetime from typing import Optional

@dataclass class User: """Core user entity - no framework dependencies.""" id: str email: str name: str created_at: datetime is_active: bool = True

def deactivate(self):
    """Business rule: deactivating user."""
    self.is_active = False

def can_place_order(self) -> bool:
    """Business rule: active users can order."""
    return self.is_active

domain/interfaces/user_repository.py

from abc import ABC, abstractmethod from typing import Optional, List from domain.entities.user import User

class IUserRepository(ABC): """Port: defines contract, no implementation."""

@abstractmethod
async def find_by_id(self, user_id: str) -> Optional[User]:
    pass

@abstractmethod
async def find_by_email(self, email: str) -> Optional[User]:
    pass

@abstractmethod
async def save(self, user: User) -> User:
    pass

@abstractmethod
async def delete(self, user_id: str) -> bool:
    pass

use_cases/create_user.py

from domain.entities.user import User from domain.interfaces.user_repository import IUserRepository from dataclasses import dataclass from datetime import datetime import uuid

@dataclass class CreateUserRequest: email: str name: str

@dataclass class CreateUserResponse: user: 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

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 already exists"
        )

    # Create entity
    user = User(
        id=str(uuid.uuid4()),
        email=request.email,
        name=request.name,
        created_at=datetime.now(),
        is_active=True
    )

    # Persist
    saved_user = await self.user_repository.save(user)

    return CreateUserResponse(
        user=saved_user,
        success=True
    )

adapters/repositories/postgres_user_repository.py

from domain.interfaces.user_repository import IUserRepository from domain.entities.user import User from typing import Optional import asyncpg

class PostgresUserRepository(IUserRepository): """Adapter: PostgreSQL implementation."""

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 find_by_email(self, email: str) -> Optional[User]:
    async with self.pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT * FROM users WHERE email = $1", email
        )
        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

async def delete(self, user_id: str) -> bool:
    async with self.pool.acquire() as conn:
        result = await conn.execute(
            "DELETE FROM users WHERE id = $1", user_id
        )
        return result == "DELETE 1"

def _to_entity(self, row) -> User:
    """Map database row to entity."""
    return User(
        id=row["id"],
        email=row["email"],
        name=row["name"],
        created_at=row["created_at"],
        is_active=row["is_active"]
    )

adapters/controllers/user_controller.py

from fastapi import APIRouter, Depends, HTTPException from use_cases.create_user import CreateUserUseCase, CreateUserRequest from pydantic import BaseModel

router = APIRouter()

class CreateUserDTO(BaseModel): email: str name: str

@router.post("/users") async def create_user( dto: CreateUserDTO, use_case: CreateUserUseCase = Depends(get_create_user_use_case) ): """Controller: handles HTTP concerns only.""" request = CreateUserRequest(email=dto.email, name=dto.name) response = await use_case.execute(request)

if not response.success:
    raise HTTPException(status_code=400, detail=response.error)

return {"user": response.user}

Hexagonal Architecture Pattern

Core domain (hexagon center)

class OrderService: """Domain service - no infrastructure dependencies."""

def __init__(
    self,
    order_repository: OrderRepositoryPort,
    payment_gateway: PaymentGatewayPort,
    notification_service: NotificationPort
):
    self.orders = order_repository
    self.payments = payment_gateway
    self.notifications = notification_service

async def place_order(self, order: Order) -> OrderResult:
    # Business logic
    if not order.is_valid():
        return OrderResult(success=False, error="Invalid order")

    # Use ports (interfaces)
    payment = await self.payments.charge(
        amount=order.total,
        customer=order.customer_id
    )

    if not payment.success:
        return OrderResult(success=False, error="Payment failed")

    order.mark_as_paid()
    saved_order = await self.orders.save(order)

    await self.notifications.send(
        to=order.customer_email,
        subject="Order confirmed",
        body=f"Order {order.id} confirmed"
    )

    return OrderResult(success=True, order=saved_order)

Ports (interfaces)

class OrderRepositoryPort(ABC): @abstractmethod async def save(self, order: Order) -> Order: pass

class PaymentGatewayPort(ABC): @abstractmethod async def charge(self, amount: Money, customer: str) -> PaymentResult: pass

class NotificationPort(ABC): @abstractmethod async def send(self, to: str, subject: str, body: str): pass

Adapters (implementations)

class StripePaymentAdapter(PaymentGatewayPort): """Primary adapter: connects to Stripe API."""

def __init__(self, api_key: str):
    self.stripe = stripe
    self.stripe.api_key = api_key

async def charge(self, amount: Money, customer: str) -> PaymentResult:
    try:
        charge = self.stripe.Charge.create(
            amount=amount.cents,
            currency=amount.currency,
            customer=customer
        )
        return PaymentResult(success=True, transaction_id=charge.id)
    except stripe.error.CardError as e:
        return PaymentResult(success=False, error=str(e))

class MockPaymentAdapter(PaymentGatewayPort): """Test adapter: no external dependencies."""

async def charge(self, amount: Money, customer: str) -> PaymentResult:
    return PaymentResult(success=True, transaction_id="mock-123")

Domain-Driven Design Pattern

Value Objects (immutable)

from dataclasses import dataclass from typing import Optional

@dataclass(frozen=True) class Email: """Value object: validated email.""" value: str

def __post_init__(self):
    if "@" not in self.value:
        raise ValueError("Invalid email")

@dataclass(frozen=True) class Money: """Value object: amount with currency.""" 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)

Entities (with identity)

class Order: """Entity: has identity, mutable state."""

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 entity."""
    item = OrderItem(product, quantity)
    self.items.append(item)
    self._events.append(ItemAddedEvent(self.id, item))

def total(self) -> Money:
    """Calculated property."""
    return sum(item.subtotal() for item in self.items)

def submit(self):
    """State transition with business rules."""
    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))

Aggregates (consistency boundary)

class Customer: """Aggregate root: controls access to entities."""

def __init__(self, id: str, email: Email):
    self.id = id
    self.email = email
    self._addresses: List[Address] = []
    self._orders: List[str] = []  # Order IDs, not full objects

def add_address(self, address: Address):
    """Aggregate enforces invariants."""
    if len(self._addresses) >= 5:
        raise ValueError("Maximum 5 addresses allowed")
    self._addresses.append(address)

@property
def primary_address(self) -> Optional[Address]:
    return next((a for a in self._addresses if a.is_primary), None)

Domain Events

@dataclass class OrderSubmittedEvent: order_id: str occurred_at: datetime = field(default_factory=datetime.now)

Repository (aggregate persistence)

class OrderRepository: """Repository: persist/retrieve aggregates."""

async def find_by_id(self, order_id: str) -> Optional[Order]:
    """Reconstitute aggregate from storage."""
    pass

async def save(self, order: Order):
    """Persist aggregate and publish events."""
    await self._persist(order)
    await self._publish_events(order._events)
    order._events.clear()

Resources

• references/clean-architecture-guide.md: Detailed layer breakdown

• references/hexagonal-architecture-guide.md: Ports and adapters patterns

• references/ddd-tactical-patterns.md: Entities, value objects, aggregates

• assets/clean-architecture-template/: Complete project structure

• assets/ddd-examples/: Domain modeling examples

Best Practices

• Dependency Rule: Dependencies always point inward

• Interface Segregation: Small, focused interfaces

• Business Logic in Domain: Keep frameworks out of core

• Test Independence: Core testable without infrastructure

• Bounded Contexts: Clear domain boundaries

• Ubiquitous Language: Consistent terminology

• Thin Controllers: Delegate to use cases

• Rich Domain Models: Behavior with data

Common Pitfalls

• Anemic Domain: Entities with only data, no behavior

• Framework Coupling: Business logic depends on frameworks

• Fat Controllers: Business logic in controllers

• Repository Leakage: Exposing ORM objects

• Missing Abstractions: Concrete dependencies in core

• Over-Engineering: Clean architecture for simple CRUD