hexagonal-architecture

name	hexagonal-architecture
description	Hexagonal architecture patterns and implementation
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"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

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 .currency != other.currency:
             ValueError()
         Money(.amount + other.amount, .currency)



 :
    
    
     ():
        . =   (uuid.uuid4())
        ._email = email
        .name = name
        .created_at = created_at  datetime.utcnow()
        ._events: [DomainEvent] = []
    

     () -> Email:
         ._email
    
     () -> :
        
         new_email == ._email:
            
        
        ._email = new_email
        ._events.append(UserEmailChanged(., new_email))
    
     () -> [DomainEvent]:
        
        events = ._events.copy()
        ._events.clear()
         events



 :
    
    
     ():
        .aggregate_id = aggregate_id
        .occurred_at = datetime.utcnow()


 ():
    
    
     ():
        ().__init__(user_id)
        .email = email
        .name = name


 ():
    
    
     ():
        ().__init__(user_id)
        .new_email = new_email



 :
    
    
     ():
        . =   (uuid.uuid4())
        .user = user
        .items = items
        .status = OrderStatus.DRAFT
        .created_at = datetime.utcnow()
        ._events: [DomainEvent] = []
    

     () -> Money:
        
        total = (item.price.amount  item  .items)
         Money(total, )
    
     () -> :
        
          .items:
             ValueError()
        
        .status = OrderStatus.SUBMITTED
        ._events.append(OrderSubmitted(.))
    
     () -> [DomainEvent]:
        events = ._events.copy()
        ._events.clear()
         events



 :
    
    product_id: 
    name: 
    quantity: 
    price: Money


 :
    
    DRAFT = 
    SUBMITTED = 
    PAID = 
    SHIPPED = 
    DELIVERED = 
    CANCELLED = 

Ports (Interfaces)

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 () -> [User]:
        
        
    

     () -> :
        
        



 ():
    
    

     () -> :
        
        



 ():
    
    

     () -> :
        
        
    

     () -> :
        
        

Application Service

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 = .user_repository.save(user)
        
        
        events = saved_user.pull_events()
        
        
        .event_publisher.publish_all(events)
        
        
        .email_service.send_email(
            to=email,
            subject=,
            body=
        )
        
         saved_user
    
     () -> User:
        
        
        user = .user_repository.find_by_id(.user_id)
          user:
             ValueError()
        
        
        new_email = Email(.new_email)
         .user_repository.exists_by_email(new_email):
             ValueError()
        
        
        user.change_email(new_email)
        
        
        saved_user = .user_repository.save(user)
        
        
        events = saved_user.pull_events()
        .event_publisher.publish_all(events)
        
         saved_user
    
     () -> [User]:
        
         .user_repository.find_by_id(user_id)

Adapters

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."""
         infrastructure.persistence.user_entity  UserEntity
        
        count = .session.query(UserEntity).(
            UserEntity.email == email.value
        ).count()
        
         count > 



 ():
    
    
     ():
        .smtp_host = smtp_host
        .smtp_port = smtp_port
    
     () -> :
        
         smtplib
         email.mime.text  MIMEText
        
        msg = MIMEText(body)
        msg[] = subject
        msg[] = 
        msg[] = to.value
        
         smtplib.SMTP(.smtp_host, .smtp_port)  server:
            server.send_message(msg)



 fastapi  FastAPI, Depends


app = FastAPI()


 () -> UserApplicationService:
    
    
     main  container
     container.resolve(UserApplicationService)



 ():
    
    :
        user = service.create_user()
         {: user., : user.email.value}
     ValueError  e:
         {: (e)}, 

Dependency Injection Setup

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,
    )


# Bootstrap
def create_container() -> Container:
    container = Container()
    container.config.from_yaml("config.yaml")
    return container

Best Practices

1. Keep domain pure
   No dependencies on infrastructure
   No I/O in domain models

2. Define clear ports
   Separate inbound (use cases) from outbound (persistence)

3. Invert dependencies
   Domain depends on abstractions
   Infrastructure depends on domain

4. Aggregate design
   One aggregate root per transaction
   References by ID only

5. Domain events
   Capture state changes as events
   Decouple through events

6. Application services
   Orchestrate domain objects
   Handle transactions

7. Testability
   Mock all outbound ports
   Test domain logic in isolation

8. Single responsibility
   Each port has one purpose
   Each adapter implements one port