--- name: architecture-patterns description: 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. ---

# 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

```python # 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

```python # 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

```python # 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, i

--- name: architecture-patterns description: 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. ---

# 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

```python # 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

```python # 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

```python # 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, i

--- name: architecture-patterns description: 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. ---

# 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

```python # 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

```python # 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

```python # 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, i

--- name: architecture-patterns description: 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. ---

# 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

```python # 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

```python # 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

```python # 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, i