hexagonal-architecture

## What I do - Design ports and adapters architecture - Implement domain-driven design - Create dependency inversion layers - Structure applications for testability - Handle external dependencies - Design clear boundaries - Implement application services - Manage domain models ## When to use me When designing application architecture or refactoring to hexagonal architecture. ## Hexagonal Architecture Overview ``` ┌─────────────────────────────────────────────────────────────────┐ │ External Layers │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Web/API │ │ CLI │ │ Tests │ │ │ │ Adapters │ │ Adapter │ │ Adapters │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ └────────────────┼────────────────┘ │ │ │ │ │ ┌───────────▼───────────┐ │ │ │ Adapters │ │ │ │ (Infrastructure) │ │ │ └───────────┬───────────┘ │ │ │ │ ├──────────────────────────┼──────────────────────────────────────┤ │ Application │ Domain Layer │ │ │ │ │ ┌───────────────────────▼───────────────────────┐ │ │ │ Ports (Interfaces) │ │ │ │ │ │ │ │ Inbound Ports: │ Outbound Ports: │ │ │ │ - Use Cases │ - Repositories │ │ │ │ - Services │ - External Services │ │ │ │ │ - Event Publishers │ │ │ └───────────────────┼─────────────────────────┘ │ │ │ │ │ ┌──────────────▼──────────────┐ │ │ │ Domain Models │ │ │ │ │ │ │ │ - Entities │ │ │ │ - Value Objects │ │ │ │ - Aggregates │ │ │ │ - Domain Events │ │ │ │ - Domain Services │ │ │ └────────────────────────────┘ │ │ │ ├────────────────────────────────────────────────────────────────┤ │ External Systems │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Database │ │ Cache │ │ APIs │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ └────────────────────────────────────────────────────────────────┘ ``` ## Domain Layer ```python from dataclasses import dataclass from datetime import datetime from typing import List, Optional from abc import ABC, abstractmethod import uuid # Value Objects @dataclass(frozen=True) class Email: """Email value object.""" value: str def __post_init__(self): if '@' not in self.value: raise ValueError(f"Invalid email: {self.value}") @dataclass(frozen=True) class Money: """Money value object with currency.""" amount: float currency: str def __post_init__(self): if self.amount < 0: raise ValueError("Amount cannot be negative") def add(self, other: 'Money') -> 'Money': if self.currency != other.currency: raise ValueError("Cannot add different currencies") return Money(self.amount + other.amount, self.currency) # Entities class User: """User entity with identity and behavior.""" def __init__( self, email: Email, name: str, id: str = None, created_at: datetime = None, ): self.id = id or str(uuid.uuid4()) self._email = email self.name = name self.created_at = created_at or datetime.utcnow() self._events: List[DomainEvent] = [] @property def email(self) -> Email: return self._email def change_email(self, new_email: Email) -> None: """Change email with validation.""" if new_email == self._email: return self._email = new_email self._events.append(UserEmailChanged(self.id, new_email)) def pull_events(self) -> List[DomainEvent]: """Get domain events.""" events = self._events.copy() self._events.clear() return events # Domain Events class DomainEvent: """Base domain event.""" def __init__(self, aggregate_id: str): self.aggregate_id = aggregate_id self.occurred_at = datetime.utcnow() class UserCreated(DomainEvent): """User was created.""" def __init__(self, user_id: str, email: Email, name: str): super().__init__(user_id) self.email = email self.name = name class UserEmailChanged(DomainEvent): """User email was changed.""" def __init__(self, user_id: str, new_email: Email): super().__init__(user_id) self.new_email = new_email # Aggregates class Order: """Order aggregate root.""" def __init__( self, user: User, items: List['OrderItem'], id: str = None, ): self.id = id or str(uuid.uuid4()) self.user = user self.items = items self.status = OrderStatus.DRAFT self.created_at = datetime.utcnow() self._events: List[DomainEvent] = [] @property def total(self) -> Money: """Calculate order total.""" total = sum(item.price.amount for item in self.items) return Money(total, 'USD') def submit(self) -> None: """Submit order.""" if not self.items: raise ValueError("Cannot submit empty order") self.status = OrderStatus.SUBMITTED self._events.append(OrderSubmitted(self.id)) def pull_events(self) -> List[DomainEvent]: events = self._events.copy() self._events.clear() return events @dataclass class OrderItem: """Order item value object.""" product_id: str name: str quantity: int price: Money class OrderStatus: """Order status enum.""" DRAFT = "draft" SUBMITTED = "submitted" PAID = "paid" SHIPPED = "shipped" DELIVERED = "delivered" CANCELLED = "cancelled" ``` ## Ports (Interfaces) ```python from abc import ABC from typing import List, Optional # Inbound Ports (Application Services) class UserServicePort(ABC): """Inbound port for user operations.""" @abstractmethod def create_user(self, email: str, name: str) -> User: """Create a new user.""" pass @abstractmethod def get_user(self, user_id: str) -> Optional[User]: """Get user by ID.""" pass @abstractmethod def change_user_email( self, user_id: str, new_email: str ) -> User: """Change user email.""" pass # Outbound Ports (Repositories) class UserRepositoryPort(ABC): """Outbound port for user persistence.""" @abstractmethod def save(self, user: User) -> User: """Save user.""" pass @abstractmethod def find_by_id(self, user_id: str) -> Optional[User]: """Find user by ID.""" pass @abstractmethod def exists_by_email(self, email: Email) -> bool: """Check if email exists.""" pass # Outbound Ports (External Services) class EmailServicePort(ABC): """Outbound port for sending emails.""" @abstractmethod def send_email( self, to: Email, subject: str, body: str ) -> None: """Send email.""" pass # Outbound Ports (Event Publishing) class EventPublisherPort(ABC): """Outbound port for publishing domain events.""" @abstractmethod def publish(self, event: DomainEvent) -> None: """Publish domain event.""" pass @abstractmethod def publish_all(self, events: List[DomainEvent]) -> None: """Publish multiple events.""" pass ``` ## Application Service ```python from dataclasses import dataclass @dataclass class CreateUserInput: """Input for creating a user.""" email: str name: str @dataclass class ChangeEmailInput: """Input for changing email.""" user_id: str new_email: str class UserApplicationService: """ Application service implementing UserServicePort. Coordinates between inbound ports (use cases) and outbound ports (infrastructure). """ def __init__( self, user_repository: UserRepositoryPort, email_service: EmailServicePort, event_publisher: EventPublisherPort, ): self.user_repository = user_repository self.email_service = email_service self.event_publisher = event_publisher def create_user(self, input: CreateUserInput) -> User: """Create a new user.""" # Validate email doesn't exist email = Email(input.email) if self.user_repository.exists_by_email(email): raise ValueError(f"Email already exists: {input.email}") # Create user (domain logic) user = User( email=email, name=input.name, ) # Save to persistence (through port) saved_user = self.user_repository.save(user) # Get domain events events = saved_user.pull_events() # Publish events (through port) self.event_publisher.publish_all(events) # Send welcome email (through port) self.email_service.send_email( to=email, subject="Welcome!", body=f"Hi {input.name}, welcome to our platform!" ) return saved_user def change_user_email( self, input: ChangeEmailInput ) -> User: """Change user email.""" # Get user user = self.user_repository.find_by_id(input.user_id) if not user: raise ValueError(f"User not found: {input.user_id}") # Validate new email new_email = Email(input.new_email) if self.user_repository.exists_by_email(new_email): raise ValueError(f"Email already exists: {input.new_email}") # Change email (domain logic) user.change_email(new_email) # Save saved_user = self.user_repository.save(user) # Publish events events = saved_user.pull_events() self.event_publisher.publish_all(events) return saved_user def get_user(self, user_id: str) -> Optional[User]: """Get user by ID.""" return self.user_repository.find_by_id(user_id) ``` ## Adapters ```python from sqlalchemy.orm import Session # Database Adapter (Implements UserRepositoryPort) class SQLUserRepository(UserRepositoryPort): """SQLAlchemy implementation of user repository.""" def __init__(self, session: Session): self.session = session def save(self, user: User) -> User: """Save user to database.""" from infrastructure.persistence.user_entity import UserEntity # Convert domain to entity entity = UserEntity.from_domain(user) self.session.add(entity) self.session.commit() self.session.refresh(entity) return entity.to_domain() def find_by_id(self, user_id: str) -> Optional[User]: """Find user by ID.""" from infrastructure.persistence.user_entity import UserEntity entity = self.session.query(UserEntity).filter( UserEntity.id == user_id ).first() return entity.to_domain() if entity else None def exists_by_email(self, email: Email) -> bool: """Check if email exists.""" from infrastructure.persistence.user_entity import UserEntity count = self.session.query(UserEntity).filter( UserEntity.email == email.value ).count() return count > 0 # Email Adapter (Implements EmailServicePort) class SMTPEmailService(EmailServicePort): """SMTP implementation of email service.""" def __init__(self, smtp_host: str, smtp_port: int): self.smtp_host = smtp_host self.smtp_port = smtp_port def send_email( self, to: Email, subject: str, body: str ) -> None: """Send email via SMTP.""" import smtplib from email.mime.text import MIMEText msg = MIMEText(body) msg['Subject'] = subject msg['From'] = "noreply@example.com" msg['To'] = to.value with smtplib.SMTP(self.smtp_host, self.smtp_port) as server: server.send_message(msg) # HTTP Adapter (Inbound Adapter) from fastapi import FastAPI, Depends app = FastAPI() def get_user_service() -> UserApplicationService: """Get user service from DI container.""" # In production, use proper DI container from main import container return container.resolve(UserApplicationService) @app.post("/users") def create_user( input: CreateUserInput, service: UserApplicationService = Depends(get_user_service) ): """Create user via HTTP.""" try: user = service.create_user(input) return {"id": user.id, "email": user.email.value} except ValueError as e: return {"error": str(e)}, 400 ``` ## Dependency Injection Setup ```python from dependency_injector import containers, providers class Container(containers.DeclarativeContainer): """DI container for the application.""" # Configuration config = providers.Configuration() # Database database_session = providers.Singleton( create_database_session, url=config.database.url, ) # Repositories user_repository = providers.Factory( SQLUserRepository, session=database_session, ) # Services email_service = providers.Singleton( SMTPEmailService, smtp_host=config.email.smtp_host, smtp_port=config.email.smtp_port, ) event_publisher = providers.Singleton( KafkaEventPublisher, bootstrap_servers=config.kafka.servers, ) # Application Services user_service = providers.Factory( UserApplicationService, user_repository=user_repository, email_service=email_service, event_publisher=event_publisher,