Implement proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design. Use when architecting complex backend systems or refactoring existing applications for better maintainability.
/plugin marketplace add secondsky/claude-skills/plugin install architecture-patterns@claude-skillsThis skill inherits all available tools. When active, it can use any tool Claude has access to.
Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems.
Layers (dependency flows inward):
Key Principles:
Components:
Benefits:
Strategic Patterns:
Tactical Patterns:
app/
├── domain/ # Entities & business rules
│ ├── entities/
│ ├── value_objects/
│ └── interfaces/ # Abstract interfaces
├── use_cases/ # Application business rules
├── adapters/ # Interface implementations
│ ├── repositories/
│ ├── controllers/
│ └── gateways/
└── infrastructure/ # Framework & external concerns
# domain/entities/user.py
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):
"""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
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_cases/create_user.py
from dataclasses import dataclass
import uuid
from datetime import datetime
@dataclass
class CreateUserRequest:
email: str
name: str
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
class PostgresUserRepository(IUserRepository):
"""Adapter: PostgreSQL implementation."""
def __init__(self, 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
# 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)
# Adapters
class StripePaymentAdapter(PaymentGatewayPort):
"""Primary adapter: connects to Stripe API."""
async def charge(self, amount: Money, customer: str) -> PaymentResult:
# Implementation
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")
# Value Objects (immutable)
@dataclass(frozen=True)
class Email:
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)
# Entities (with identity)
class Order:
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):
item = OrderItem(product, quantity)
self.items.append(item)
self._events.append(ItemAddedEvent(self.id, item))
def submit(self):
if not self.items:
raise ValueError("Cannot submit empty order")
self.status = OrderStatus.SUBMITTED
self._events.append(OrderSubmittedEvent(self.id))
# Aggregates (consistency boundary)
class Customer:
def __init__(self, id: str, email: Email):
self.id = id
self.email = email
self._addresses: List[Address] = []
def add_address(self, address: Address):
if len(self._addresses) >= 5:
raise ValueError("Maximum 5 addresses allowed")
self._addresses.append(address)