hexagonal-architecture

name	Hexagonal Architecture
description	Port and adapter pattern for separating core business logic from external concerns
category	software-development

Hexagonal Architecture

What I do

I provide a layered architecture that separates application core from external dependencies. Hexagonal architecture (Ports and Adapters) puts business logic at the center, with ports defining interfaces for incoming and outgoing interactions. Adapters implement these ports, translating between external formats and internal domain objects. This creates a system that is flexible, testable, and independent of frameworks, databases, or external services.

When to use me

Use hexagonal architecture for applications that need to be flexible about their dependencies, when you want to isolate business logic from infrastructure, or when different adapters might be needed for different deployment scenarios. It's ideal for long-lived applications where frameworks or databases might change. Perfect for testing scenarios where you want to mock external dependencies completely. Avoid for simple scripts or throwaway prototypes.

Core Concepts

• Domain/Core: Pure business logic with no dependencies
• Ports: Interfaces defining inbound/outbound contracts
• Adapters: Implementations of ports for specific technologies
• Primary/Driving Adapters: Initiate actions (HTTP, CLI, GUI)
• Secondary/Driven Adapters: Handle outputs (DB, email, external APIs)
• Dependency Inversion: Core depends on abstractions, not implementations
• Application Services: Use cases coordinating domain objects
• Domain Services: Business logic spanning multiple entities
• DTOs: Data transfer objects for adapter boundaries
• Mappers: Convert between DTOs and domain objects

Code Examples

Core Domain

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol
from uuid import UUID, uuid4

class User:
    def __init__(self, user_id: UUID, email: str, name: str):
        self._id = user_id
        self._email = email
        self._name = name
        self._created_at = datetime.utcnow()
        self._is_active = True
    
    @property
    def id(self) -> UUID:
        return self._id
    
    @property
    def email(self) -> str:
        return self._email
    
    @property
    def name(self) -> str:
        return self._name
    
    @property
    def is_active(self) -> bool:
         ._is_active
    
     () -> :
        ._is_active = 
    
     () -> :
        ._is_active = 

 ():

     () -> :
        
    

     () -> User | :
        
    

     () -> User | :
        
    

     () -> [User]:
        

 :
     ():
        ._repository = user_repository
    
     () -> User:
         ._repository.find_by_email(email):
             ValueError()
        
        user = User(user_id=uuid4(), email=email, name=name)
        ._repository.save(user)
         user
    
     () -> User:
        user = ._repository.find_by_id(user_id)
          user:
             ValueError()
        user.activate()
        ._repository.save(user)
         user
    
     () -> [User]:
         ._repository.find_all_active()

Ports (Interfaces)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol
from uuid import UUID

# Inbound Ports (Application Services)
class CreateUserUseCase(Protocol):
    def execute(self, command: 'CreateUserCommand') -> 'UserResponse':
        pass

class GetUserUseCase(Protocol):
    def execute(self, user_id: UUID) -> 'UserResponse | None':
        pass

class ListUsersUseCase(Protocol):
    def execute(self) -> list['UserResponse']:
        pass

# Outbound Ports (Infrastructure Contracts)
class UserPersistencePort(Protocol):
    @abstractmethod
    def save(self, user: User) -> None:
        pass
    
    @abstractmethod
    def () -> User | :
        
    

     () -> User | :
        
    

     () -> [User]:
        

 ():

     () -> :
        
    

     () -> :
        



 :
    email: 
    name: 
    send_welcome_email:  = 


 :
    user_id: UUID
    email: 
    name: 
    created_at: datetime
    is_active: 

Application Services

class UserApplicationService:
    def __init__(
        self,
        user_repository: UserPersistencePort,
        email_service: EmailNotificationPort
    ):
        self._repository = user_repository
        self._email = email_service
    
    def create_user(self, command: CreateUserCommand) -> UserResponse:
        if self._repository.find_by_email(command.email):
            raise ValueError(f"Email {command.email} already exists")
        
        user = User(
            user_id=uuid4(),
            email=command.email,
            name=command.name
        )
        self._repository.save(user)
        
        if command.send_welcome_email:
            self._email.send_welcome_email(user)
        
        return self._to_response(user)
    
    def get_user(self, user_id: UUID) -> UserResponse | None:
        user = self._repository.find_by_id(user_id)
        if not user:
            return None
        return self._to_response(user)
    
    def list_users(self) -> list[UserResponse]:
        users = self._repository.find_all_active()
        return [self._to_response(u) for u in users]
    
     () -> UserResponse:
        user = ._repository.find_by_id(user_id)
          user:
             ValueError()
        
        user.activate()
        ._repository.save(user)
        ._email.send_activation_email(user)
        
         ._to_response(user)
    
     () -> UserResponse:
         UserResponse(
            user_id=user.,
            email=user.email,
            name=user.name,
            created_at=user._created_at,
            is_active=user.is_active
        )

Adapters (Infrastructure)

from dataclasses import dataclass
from datetime import datetime
from typing import Protocol, Optional
from uuid import UUID, uuid4

# In-Memory Adapter
class InMemoryUserRepository(UserPersistencePort):
    def __init__(self):
        self._users: dict[UUID, User] = {}
    
    def save(self, user: User) -> None:
        self._users[user.id] = user
    
    def find_by_id(self, user_id: UUID) -> Optional[User]:
        return self._users.get(user_id)
    
    def find_by_email(self, email: str) -> Optional[User]:
        for user in self._users.values():
            if user.email == email:
                return user
        return None
    
    def find_all_active(self) -> list[User]:
        return [u for u in ._users.values()  u.is_active]


 ():
     () -> :
        ()
    
     () -> :
        ()


 ():
     ():
        ._session_factory = session_factory
    
     () -> :
        session = ._session_factory()
        session.add(._to_orm(user))
        session.commit()
    
     () -> [User]:
        session = ._session_factory()
        orm_user = session.query(UserORM).filter_by(=user_id).first()
         ._from_orm(orm_user)  orm_user  
    
     () -> [User]:
        session = ._session_factory()
        orm_user = session.query(UserORM).filter_by(email=email).first()
         ._from_orm(orm_user)  orm_user  
    
     () -> [User]:
        session = ._session_factory()
        orm_users = session.query(UserORM).filter_by(is_active=).()
         [._from_orm(u)  u  orm_users]
    
     () -> :
         UserORM(
            =user.,
            email=user.email,
            name=user.name,
            is_active=user.is_active
        )
    
     () -> User:
         User(
            user_id=orm_user.,
            email=orm_user.email,
            name=orm_user.name
        )


 :
    : UUID
    email: 
    name: 
    is_active: 

HTTP Adapter (FastAPI)

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr
from typing import list
from uuid import UUID

app = FastAPI(title="User Service API")

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str
    send_welcome_email: bool = True

class UserResponse(BaseModel):
    user_id: UUID
    email: EmailStr
    name: str
    is_active: bool

def get_user_service():
    from main import container
    return container.resolve(UserApplicationService)

@app.post("/users", response_model=UserResponse)
async def create_user(
    request: CreateUserRequest,
    service=Depends(get_user_service)
):
    try:
        command = CreateUserCommand(
            email=request.email,
            name=request.name,
            send_welcome_email=request.send_welcome_email
        )
        return service.create_user(command)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/users/{user_id}", response_model=UserResponse)
  ():
    user = service.get_user(user_id)
      user:
         HTTPException(status_code=, detail=)
     user


  ():
     service.list_users()


  ():
    :
         service.activate_user(user_id)
     ValueError  e:
         HTTPException(status_code=, detail=(e))

Dependency Injection Container

from typing import Protocol, Type, TypeVar, get_origin, get_args

T = TypeVar("T")

class Container:
    def __init__(self):
        self._bindings: dict[Type, object] = {}
        self._singletons: dict[Type, object] = {}
    
    def bind(self, abstract: Type[T], concrete: Type[T]) -> None:
        self._bindings[abstract] = concrete
    
    def singleton(self, abstract: Type[T], concrete: Type[T]) -> None:
        self._singletons[abstract] = None
        self._bindings[abstract] = concrete
    
    def resolve(self, abstract: Type[T]) -> T:
        if abstract in self._singletons:
            if self._singletons[abstract] is None:
                self._singletons[abstract] = self._build(abstract)
            return self._singletons[abstract]
         ._build(abstract)
    
     () -> T:
         abstract   ._bindings:
              (abstract, ):
                 abstract
             ValueError()
        
        concrete = ._bindings[abstract]
        
         (concrete, ):
            params = ._get_init_params(concrete)
            deps = {param: .resolve(param_type)  param, param_type  params.items()}
             concrete(**deps)
        
         concrete
    
     () -> [, ]:
         inspect
        sig = inspect.signature(cls.__init__)
         {
            name: param.annotation
             name, param  sig.parameters.items()
             name !=   param.annotation != inspect.Parameter.empty
        }


container = Container()
container.singleton(UserPersistencePort, InMemoryUserRepository)
container.singleton(EmailNotificationPort, ConsoleEmailAdapter)
container.singleton(UserApplicationService, UserApplicationService)


 ():
     container

 ():
     container.resolve(UserApplicationService)

Best Practices

1. Keep Core Pure: Domain should have no external dependencies
2. Define Clear Ports: Interfaces should be focused and small
3. Adapter per Concern: Separate adapters for different technologies
4. Use Dependency Injection: Let framework assemble components
5. Test Core in Isolation: Mock all ports for domain tests
6. Test Adapters Separately: Integration tests for each adapter
7. Configure in Entry Point: Setup bindings in main/application
8. Avoid Leaking Abstractions: Ports should be in core, not adapters
9. Multiple Implementations: Support different environments (dev/test/prod)
10. Clean Boundaries: DTOs for all external communication
11. Frameagnostic Core: Core should not import framework code
12. Evolve Ports Carefully: Changing ports affects all adapters