Architecture Patterns
Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems.
Given: a service boundary or module to architect.
Produces: layered structure with clear dependency rules, interface definitions, and test boundaries.
When to Use This Skill
- Designing new backend services or microservices from scratch
- Refactoring monolithic applications where business logic is entangled with ORM models or HTTP concerns
- Establishing bounded contexts before splitting a system into services
- Debugging dependency cycles where infrastructure code bleeds into the domain layer
- Creating testable codebases where use-case tests do not require a running database
- Implementing domain-driven design tactical patterns (aggregates, value objects, domain events)
Core Concepts
1. Clean Architecture (Uncle Bob)
Layers (dependency flows inward):
- Entities: Core business models, no framework imports
- Use Cases: Application business rules, orchestrate entities
- Interface Adapters: Controllers, presenters, gateways β translate between use cases and external formats
- Frameworks & Drivers: UI, database, external services β all at the outermost ring
Key Principles:
- Dependencies point inward only; inner layers know nothing about outer layers
- Business logic is independent of frameworks, databases, and delivery mechanisms
- Every layer boundary is crossed via an abstract interface
- Testable without UI, database, or external services
2. Hexagonal Architecture (Ports and Adapters)
Components:
- Domain Core: Business logic lives here, framework-free
- Ports: Abstract interfaces that define how the core interacts with the outside world (driving and driven)
- Adapters: Concrete implementations of ports (PostgreSQL adapter, Stripe adapter, REST adapter)
Benefits:
- Swap implementations without touching the core (e.g., replace PostgreSQL with DynamoDB)
- Use in-memory adapters in tests β no Docker required
- Technology decisions deferred to the edges
3. Domain-Driven Design (DDD)
Strategic Patterns:
- Bounded Contexts: Isolate a coherent model for one subdomain; avoid sharing a single model across the whole system
- Context Mapping: Define how contexts relate (Anti-Corruption Layer, Shared Kernel, Open Host Service)
- Ubiquitous Language: Every term in code matches the term used by domain experts
Tactical Patterns:
- Entities: Objects with stable identity that change over time
- Value Objects: Immutable objects identified by their attributes (Email, Money, Address)
- Aggregates: Consistency boundaries; only the root is accessible from outside
- Repositories: Persist and reconstitute aggregates; abstract over the storage mechanism
- Domain Events: Capture things that happened inside the domain; used for cross-aggregate coordination
Clean Architecture β Directory Structure
app/
βββ domain/ # Entities, value objects, interfaces
β βββ entities/
β β βββ user.py
β β βββ order.py
β βββ value_objects/
β β βββ email.py
β β βββ money.py
β βββ interfaces/ # Abstract ports (no implementations)
β βββ user_repository.py
β βββ payment_gateway.py
βββ use_cases/ # Application business rules
β βββ create_user.py
β βββ process_order.py
β βββ send_notification.py
βββ adapters/ # Concrete 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 wiring, config, DI container
βββ database.py
βββ config.py
βββ logging.py
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/.
Clean Architecture β Core Implementation
from dataclasses import dataclass
from datetime import datetime
@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):
self.is_active = False
def can_place_order(self) -> bool:
return self.is_active
from abc import ABC, abstractmethod
from typing import Optional
from domain.entities.user import User
class IUserRepository(ABC):
"""Port: defines contract, no implementation details."""
@abstractmethod
async def find_by_id(self, user_id: str) -> Optional[User]: ...
@abstractmethod
async def find_by_email(self, email: str) -> Optional[User]: ...
@abstractmethod
async def save(self, user: User) -> User: ...
@abstractmethod
async def delete(self, user_id: str) -> bool: ...
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import uuid
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository
@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, no HTTP or DB details."""
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository
async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
existing = await self.user_repository.find_by_email(request.email)
if existing:
return CreateUserResponse(user=None, success=False, error="Email already exists")
user = User(
id=str(uuid.uuid4()),
email=request.email,
name=request.name,
created_at=datetime.now(),
)
saved_user = await self.user_repository.save(user)
return CreateUserResponse(user=saved_user, success=True)
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 of the user 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 find_by_email(self, email: str) -> Optional[User]:
async with self.pool.acquire(