| name | architecture-patterns |
| description | Implement proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, Domain-Driven Design, and deep module design. Use this skill when designing clean architecture for a new service, when refactoring a monolith to use bounded contexts, when implementing hexagonal or onion architecture patterns, when debugging dependency cycles between application layers, or when making architectural decisions about module depth, seams, and interfaces. |
Architecture Patterns
Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, Domain-Driven Design, and deep module 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)
- Evaluating whether a module is deep or shallow, and whether a seam is justified
- Setting up a domain glossary (
CONTEXT.md) for a project
Module Depth (Ousterhout / Pocock)
Before choosing an architecture pattern, understand the design goal that all patterns serve: depth. Every architectural decision should make modules deeper — more capability behind simpler interfaces.
Core Vocabulary
- Module — anything with an interface and an implementation (function, class, package, slice). Scale-agnostic.
- Interface — everything a caller must know: types, invariants, error modes, ordering constraints, performance characteristics. Not just the type signature.
- Depth — leverage at the interface. A deep module provides a lot of behavior behind a small interface. A shallow module has an interface nearly as complex as its implementation.
- Seam — where an interface lives; a place behavior can be altered without editing in place.
- Adapter — a concrete thing satisfying an interface at a seam.
- Leverage — what callers get from depth: more capability per unit of interface learned.
- Locality — what maintainers get from depth: change, bugs, and knowledge concentrate in one place.
Key Principles
- Depth is a property of the interface, not the implementation. A deep module can be internally composed of small parts — they just aren't part of the interface.
- The deletion test. Imagine deleting the module. If complexity vanishes, it was a pass-through. If complexity reappears across N callers, it was earning its keep.
- The interface is the test surface. Callers and tests cross the same seam. If you need to reach past the interface to test, the module is the wrong shape.
- One adapter = hypothetical seam. Two adapters = real seam. Don't introduce a seam unless something actually varies across it.
How This Relates to Architecture Patterns
Clean Architecture, Hexagonal, and DDD all create seams. The deep module lens helps you decide which seams are justified:
- A repository interface with a Postgres adapter AND an in-memory test adapter = real seam (two adapters).
- A
UserServiceInterface with exactly one UserServiceImpl and no test fake = shallow indirection. Inline it until a second adapter is needed.
Domain Glossary Convention: CONTEXT.md
Maintain a CONTEXT.md at the project root as a glossary of domain terms — not a spec, not a scratchpad, just the shared language.
Why
- Agents and developers use consistent naming for variables, functions, and files
- Reduces verbosity (agents don't explain terms they already know)
- Survives across sessions and team members
Format
# [Project Name]
## Language
**Order**:
A customer's request to purchase items. Created when checkout completes, not when items are added to cart.
_Avoid_: purchase, transaction (use only when referring to payment processing)
**Tenderer**:
A company submitting a response to an RFP. Each tenderer submits multiple deliverables.
_Avoid_: vendor, supplier, bidder
## Relationships
- A **Tenderer** submits many **Deliverables**
- A **Deliverable** is evaluated against one **Rubric**
## Flagged ambiguities
- [Record terms that were clarified during development]
Rules
- Create
CONTEXT.md lazily — only when the first term needs capturing
- Keep it devoid of implementation details
- When a new term emerges during development, add it immediately
- When a term conflicts with existing glossary, call it out and resolve
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() 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:
return User(
id=row["id"], email=row["email"], name=row["name"],
created_at=row["created_at"], is_active=row["is_active"],
)
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
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 only — no business logic lives here."""
response = await use_case.execute(CreateUserRequest(email=dto.email, name=dto.name))
if not response.success:
raise HTTPException(status_code=400, detail=response.error)
return {"user": response.user}
Hexagonal Architecture — Ports and Adapters
class OrderService:
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:
if not order.is_valid():
return OrderResult(success=False, error="Invalid order")
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)
class OrderRepositoryPort(ABC):
@abstractmethod
async def save(self, order: Order) -> Order: ...
class PaymentGatewayPort(ABC):
@abstractmethod
async def charge(self, amount: Money, customer: str) -> PaymentResult: ...
class NotificationPort(ABC):
@abstractmethod
async def send(self, to: str, subject: str, body: str): ...
class StripePaymentAdapter(PaymentGatewayPort):
def __init__(self, api_key: str):
import stripe
stripe.api_key = api_key
self._stripe = stripe
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 self._stripe.error.CardError as e:
return PaymentResult(success=False, error=str(e))
class MockPaymentAdapter(PaymentGatewayPort):
async def charge(self, amount: Money, customer: str) -> PaymentResult:
return PaymentResult(success=True, transaction_id="mock-txn-123")
DDD — Value Objects and Aggregates
from dataclasses import dataclass
@dataclass(frozen=True)
class Email:
value: str
def __post_init__(self):
if "@" not in self.value or "." not in self.value.split("@")[-1]:
raise ValueError(f"Invalid email: {self.value}")
@dataclass(frozen=True)
class Money:
amount: int
currency: str
def __post_init__(self):
if self.amount < 0:
raise ValueError("Money amount cannot be negative")
if self.currency not in {"USD", "EUR", "GBP"}:
raise ValueError(f"Unsupported currency: {self.currency}")
def add(self, other: "Money") -> "Money":
if self.currency != other.currency:
raise ValueError("Currency mismatch")
return Money(self.amount + other.amount, self.currency)
class Order:
def __init__(self, id: str, customer_id: str):
self.id = id
self.customer_id = customer_id
self.items: list[OrderItem] = []
self.status = OrderStatus.PENDING
self._events: list[DomainEvent] = []
def add_item(self, product: Product, quantity: int):
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify a submitted order")
item = OrderItem(product=product, quantity=quantity)
self.items.append(item)
self._events.append(ItemAddedEvent(order_id=self.id, item=item))
@property
def total(self) -> Money:
totals = [item.subtotal() for item in self.items]
return sum(totals[1:], totals[0]) if totals else Money(0, "USD")
def submit(self):
if not self.items:
raise ValueError("Cannot submit an empty order")
if self.status != OrderStatus.PENDING:
raise ValueError("Order already submitted")
self.status = OrderStatus.SUBMITTED
self._events.append(OrderSubmittedEvent(order_id=self.id))
def pop_events(self) -> list[DomainEvent]:
events, self._events = self._events, []
return events
class OrderRepository(ABC):
@abstractmethod
async def find_by_id(self, order_id: str) -> Optional[Order]: ...
@abstractmethod
async def save(self, order: Order) -> None: ...
Testing — In-Memory Adapters
The 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:
import asyncio
from typing import Dict, Optional
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
class InMemoryUserRepository(IUserRepository):
def __init__(self):
self._store: Dict[str, User] = {}
async def find_by_id(self, user_id: str) -> Optional[User]:
return self._store.get(user_id)
async def find_by_email(self, email: str) -> Optional[User]:
return next((u for u in self._store.values() if u.email == email), None)
async def save(self, user: User) -> User:
self._store[user.id] = user
return user
async def delete(self, user_id: str) -> bool:
return self._store.pop(user_id, None) is not None
async def test_create_user_succeeds():
repo = InMemoryUserRepository()
use_case = CreateUserUseCase(user_repository=repo)
response = await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice"))
assert response.success
assert response.user.email == "alice@example.com"
assert response.user.id is not None
async def test_duplicate_email_rejected():
repo = InMemoryUserRepository()
use_case = CreateUserUseCase(user_repository=repo)
await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice"))
response = await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice2"))
assert not response.success
assert "already exists" in response.error
Troubleshooting
Use case tests require a running database
Business 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.
Circular imports between layers
A 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/.
Framework decorators appearing in domain entities
If 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.
All logic ending up in controllers
When 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.
Value objects raising errors too late
Validate 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.
Context bleed across bounded contexts
If 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.
Advanced Patterns
For detailed DDD bounded context mapping, full multi-service project trees, Anti-Corruption Layer implementations, and Onion Architecture comparisons, see:
Related Skills
microservices-patterns — Apply these architecture patterns when decomposing a monolith into servicescqrs-implementation — Use Clean Architecture as the structural foundation for CQRS command/query separationsaga-orchestration — Sagas require well-defined aggregate boundaries, which DDD tactical patterns provideevent-store-design — Domain events produced by aggregates feed directly into an event store
