Hexagonal Architecture, also known as Ports and Adapters Architecture, is a software design pattern that promotes loose coupling between the core business logic and external systems. This architectural pattern was introduced by Alistair Cockburn to create applications that can be equally driven by users, programs, automated tests, or batch scripts, and can be developed and tested in isolation from runtime devices and databases.
Core Concepts
The architecture gets its name from the hexagonal shape used in diagrams, though the number of sides is arbitrary. The key idea is to isolate the application’s core business logic (the “inside”) from external concerns (the “outside”) through well-defined interfaces called ports.
External Systems
|
┌─────────┐
┌────│ Adapter │────┐
│ └─────────┘ │
│ │
┌─────────┐ ┌─────────┐
┌────│ Adapter │ │ Adapter │────┐
│ └─────────┘ └─────────┘ │
│ │
│ ┌─────────────┐ │
│ ┌────│ PORT │────┐ │
│ │ └─────────────┘ │ │
│ │ │ │
│ │ ┌─────────────┐ │ │
│ │ │ APPLICATION │ │ │
│ │ │ CORE │ │ │
│ │ └─────────────┘ │ │
│ │ │ │
│ │ ┌─────────────┐ │ │
│ └────│ PORT │────┘ │
│ └─────────────┘ │
│ │
│ ┌─────────┐ ┌─────────┐ │
└────│ Adapter │ │ Adapter │────┘
└─────────┘ └─────────┘
│ │
│ ┌─────────┐ │
└────│ Adapter │────┘
└─────────┘
|
External Systems
Key Components
Application Core
The center of the hexagon contains the business logic, domain models, and use cases. This core is independent of any external technology or framework.
Ports
Ports define the interface between the application core and the outside world. They are abstract contracts that specify how the core can be used or how it can interact with external systems.
- Primary Ports (Driving Ports): Define how external actors can use the application
- Secondary Ports (Driven Ports): Define how the application can interact with external systems
Adapters
Adapters implement the ports and handle the translation between the application core and external systems. They contain the technology-specific code.
- Primary Adapters (Driving Adapters): Handle incoming requests (REST controllers, CLI interfaces)
- Secondary Adapters (Driven Adapters): Handle outgoing requests (database repositories, message queues)
Benefits
Technology Independence: The core business logic remains independent of databases, web frameworks, or external services. This makes it easier to change technologies without affecting the core functionality.
Testability: Business logic can be tested in isolation without requiring external systems. Mock implementations of ports can be used for testing.
Flexibility: Different adapters can be plugged in for different environments (development, testing, production) without changing the core logic.
Maintainability: Clear separation of concerns makes the codebase easier to understand and maintain.
Implementation Guidelines
- Keep the Core Pure: Business logic should not depend on external frameworks or libraries
- Define Clear Contracts: Ports should have well-defined interfaces with clear responsibilities
- Dependency Direction: Dependencies should always point inward toward the core
- Adapter Responsibility: Adapters should only handle translation and delegation, not business logic
The hexagonal architecture is particularly valuable in complex applications where business logic needs to be preserved and tested independently of external systems. It provides a solid foundation for building maintainable, testable, and flexible software systems.
Example 1 Explanation: E-commerce Order Processing System
This example demonstrates hexagonal architecture applied to an e-commerce order processing system. The core business logic handles order creation, inventory validation, and customer notifications.
# Example 1: E-commerce Order Processing System
# This example demonstrates hexagonal architecture for processing customer orders
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from enum import Enum
# ============================================================================
# DOMAIN MODELS (Core)
# ============================================================================
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
SHIPPED = "shipped"
DELIVERED = "delivered"
@dataclass
class Product:
id: str
name: str
price: float
stock_quantity: int
@dataclass
class OrderItem:
product_id: str
quantity: int
unit_price: float
@dataclass
class Order:
id: str
customer_id: str
items: List[OrderItem]
status: OrderStatus
total_amount: float
# ============================================================================
# PORTS (Interfaces)
# ============================================================================
# Secondary Port - for data persistence
class OrderRepository(ABC):
"""Port for order data persistence"""
@abstractmethod
def save(self, order: Order) -> Order:
"""Save an order to storage"""
pass
@abstractmethod
def find_by_id(self, order_id: str) -> Optional[Order]:
"""Find an order by its ID"""
pass
# Secondary Port - for inventory management
class InventoryService(ABC):
"""Port for inventory management"""
@abstractmethod
def check_availability(self, product_id: str, quantity: int) -> bool:
"""Check if product is available in required quantity"""
pass
@abstractmethod
def reserve_items(self, product_id: str, quantity: int) -> bool:
"""Reserve items in inventory"""
pass
# Secondary Port - for notifications
class NotificationService(ABC):
"""Port for sending notifications"""
@abstractmethod
def send_order_confirmation(self, order: Order) -> None:
"""Send order confirmation to customer"""
pass
# ============================================================================
# APPLICATION CORE (Business Logic)
# ============================================================================
class OrderService:
"""Core business logic for order processing"""
def __init__(
self,
order_repo: OrderRepository,
inventory_service: InventoryService,
notification_service: NotificationService
):
# Dependencies injected through constructor
self._order_repo = order_repo
self._inventory_service = inventory_service
self._notification_service = notification_service
def process_order(self, customer_id: str, items: List[OrderItem]) -> Order:
"""
Process a new customer order
This contains the core business logic
"""
# Validate inventory availability
for item in items:
if not self._inventory_service.check_availability(
item.product_id, item.quantity
):
raise ValueError(f"Product {item.product_id} not available")
# Calculate total amount
total_amount = sum(item.quantity * item.unit_price for item in items)
# Create order
order = Order(
id=f"ORD-{customer_id}-{len(items)}", # Simple ID generation
customer_id=customer_id,
items=items,
status=OrderStatus.PENDING,
total_amount=total_amount
)
# Reserve inventory
for item in items:
self._inventory_service.reserve_items(item.product_id, item.quantity)
# Save order
order.status = OrderStatus.CONFIRMED
saved_order = self._order_repo.save(order)
# Send confirmation
self._notification_service.send_order_confirmation(saved_order)
return saved_order
def get_order(self, order_id: str) -> Optional[Order]:
"""Retrieve an order by ID"""
return self._order_repo.find_by_id(order_id)
# ============================================================================
# ADAPTERS (Implementation of Ports)
# ============================================================================
# Secondary Adapter - In-memory database simulation
class InMemoryOrderRepository(OrderRepository):
"""Concrete implementation of OrderRepository using in-memory storage"""
def __init__(self):
self._orders: dict[str, Order] = {}
def save(self, order: Order) -> Order:
"""Save order to in-memory storage"""
self._orders[order.id] = order
print(f"Order {order.id} saved to database")
return order
def find_by_id(self, order_id: str) -> Optional[Order]:
"""Find order in in-memory storage"""
return self._orders.get(order_id)
# Secondary Adapter - Simple inventory service
class SimpleInventoryService(InventoryService):
"""Concrete implementation of InventoryService"""
def __init__(self):
# Simulate product inventory
self._inventory = {
"PROD-001": 50,
"PROD-002": 25,
"PROD-003": 10
}
def check_availability(self, product_id: str, quantity: int) -> bool:
"""Check product availability"""
available = self._inventory.get(product_id, 0)
return available >= quantity
def reserve_items(self, product_id: str, quantity: int) -> bool:
"""Reserve items by reducing inventory"""
if self.check_availability(product_id, quantity):
self._inventory[product_id] -= quantity
print(f"Reserved {quantity} units of {product_id}")
return True
return False
# Secondary Adapter - Console notification service
class ConsoleNotificationService(NotificationService):
"""Concrete implementation of NotificationService using console output"""
def send_order_confirmation(self, order: Order) -> None:
"""Send confirmation by printing to console"""
print(f"📧 Order confirmation sent to customer {order.customer_id}")
print(f"Order ID: {order.id}, Total: ${order.total_amount:.2f}")
# Primary Adapter - REST API simulation
class OrderController:
"""Primary adapter that handles incoming requests (like REST API)"""
def __init__(self, order_service: OrderService):
self._order_service = order_service
def create_order(self, customer_id: str, items_data: List[dict]) -> dict:
"""Handle HTTP POST /orders request"""
try:
# Convert request data to domain objects
items = [
OrderItem(
product_id=item["product_id"],
quantity=item["quantity"],
unit_price=item["unit_price"]
)
for item in items_data
]
# Delegate to business logic
order = self._order_service.process_order(customer_id, items)
# Return response
return {
"success": True,
"order_id": order.id,
"status": order.status.value,
"total": order.total_amount
}
except ValueError as e:
return {"success": False, "error": str(e)}
def get_order(self, order_id: str) -> dict:
"""Handle HTTP GET /orders/{id} request"""
order = self._order_service.get_order(order_id)
if order:
return {
"order_id": order.id,
"customer_id": order.customer_id,
"status": order.status.value,
"total": order.total_amount,
"items": len(order.items)
}
return {"error": "Order not found"}
# ============================================================================
# COMPOSITION ROOT (Dependency Injection)
# ============================================================================
def create_order_system() -> OrderController:
"""
Compose the application by wiring dependencies
This is where we choose which adapters to use
"""
# Create secondary adapters (infrastructure)
order_repo = InMemoryOrderRepository()
inventory_service = SimpleInventoryService()
notification_service = ConsoleNotificationService()
# Create application core with injected dependencies
order_service = OrderService(
order_repo=order_repo,
inventory_service=inventory_service,
notification_service=notification_service
)
# Create primary adapter
order_controller = OrderController(order_service)
return order_controller
# ============================================================================
# USAGE EXAMPLE
# ============================================================================
if __name__ == "__main__":
# Initialize the system
controller = create_order_system()
# Simulate API requests
print("=== Creating New Order ===")
order_data = [
{"product_id": "PROD-001", "quantity": 2, "unit_price": 29.99},
{"product_id": "PROD-002", "quantity": 1, "unit_price": 49.99}
]
result = controller.create_order("CUST-123", order_data)
print(f"Order creation result: {result}")
if result["success"]:
print("\n=== Retrieving Order ===")
order_info = controller.get_order(result["order_id"])
print(f"Order details: {order_info}")
Key Architectural Elements:
Domain Models: Order, OrderItem, Product represent the core business entities without any external dependencies.
Ports (Interfaces): OrderRepository, InventoryService, and NotificationService define contracts for external interactions without specifying implementation details.
Application Core: OrderService contains pure business logic for processing orders. It validates inventory, calculates totals, and coordinates the order workflow without knowing about databases or notification mechanisms.
Adapters: Concrete implementations like InMemoryOrderRepository and ConsoleNotificationService handle the technical details of data storage and notifications.
Dependency Injection: The create_order_system() function wires everything together, allowing easy substitution of different adapters for testing or different environments.
The business logic remains completely isolated from technical concerns, making it easy to test and modify external integrations without affecting core functionality.
Example 2 Explanation: User Authentication System
This example showcases hexagonal architecture in a user authentication system, demonstrating how security concerns can be cleanly separated from business logic.
# Example 2: User Authentication System
# This example shows hexagonal architecture for user authentication and authorization
from abc import ABC, abstractmethod
from typing import Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import hashlib
# ============================================================================
# DOMAIN MODELS (Core)
# ============================================================================
@dataclass
class User:
id: str
username: str
email: str
password_hash: str
is_active: bool
created_at: datetime
@dataclass
class AuthToken:
token: str
user_id: str
expires_at: datetime
is_valid: bool
@dataclass
class LoginRequest:
username: str
password: str
# ============================================================================
# PORTS (Interfaces)
# ============================================================================
# Secondary Port - for user data storage
class UserRepository(ABC):
"""Port for user data persistence"""
@abstractmethod
def find_by_username(self, username: str) -> Optional[User]:
"""Find user by username"""
pass
@abstractmethod
def save(self, user: User) -> User:
"""Save user to storage"""
pass
# Secondary Port - for token management
class TokenService(ABC):
"""Port for authentication token management"""
@abstractmethod
def generate_token(self, user_id: str) -> AuthToken:
"""Generate authentication token for user"""
pass
@abstractmethod
def validate_token(self, token: str) -> Optional[AuthToken]:
"""Validate and return token if valid"""
pass
@abstractmethod
def revoke_token(self, token: str) -> None:
"""Revoke/invalidate a token"""
pass
# Secondary Port - for password operations
class PasswordService(ABC):
"""Port for password hashing and verification"""
@abstractmethod
def hash_password(self, password: str) -> str:
"""Hash a plain text password"""
pass
@abstractmethod
def verify_password(self, password: str, password_hash: str) -> bool:
"""Verify password against hash"""
pass
# ============================================================================
# APPLICATION CORE (Business Logic)
# ============================================================================
class AuthenticationError(Exception):
"""Custom exception for authentication failures"""
pass
class AuthService:
"""Core business logic for user authentication"""
def __init__(
self,
user_repo: UserRepository,
token_service: TokenService,
password_service: PasswordService
):
self._user_repo = user_repo
self._token_service = token_service
self._password_service = password_service
def register_user(self, username: str, email: str, password: str) -> User:
"""
Register a new user
Core business logic for user registration
"""
# Check if user already exists
existing_user = self._user_repo.find_by_username(username)
if existing_user:
raise AuthenticationError("Username already exists")
# Validate password strength (business rule)
if len(password) < 8:
raise AuthenticationError("Password must be at least 8 characters")
# Create user with hashed password
password_hash = self._password_service.hash_password(password)
user = User(
id=f"user_{username}_{len(email)}", # Simple ID generation
username=username,
email=email,
password_hash=password_hash,
is_active=True,
created_at=datetime.now()
)
return self._user_repo.save(user)
def authenticate(self, login_request: LoginRequest) -> AuthToken:
"""
Authenticate user and return token
Core authentication business logic
"""
# Find user
user = self._user_repo.find_by_username(login_request.username)
if not user:
raise AuthenticationError("Invalid credentials")
# Check if user is active
if not user.is_active:
raise AuthenticationError("Account is deactivated")
# Verify password
if not self._password_service.verify_password(
login_request.password, user.password_hash
):
raise AuthenticationError("Invalid credentials")
# Generate and return token
return self._token_service.generate_token(user.id)
def validate_session(self, token: str) -> Optional[User]:
"""
Validate user session using token
"""
auth_token = self._token_service.validate_token(token)
if not auth_token or not auth_token.is_valid:
return None
# Find user associated with token
user = self._user_repo.find_by_username(auth_token.user_id)
return user if user and user.is_active else None
def logout(self, token: str) -> None:
"""Logout user by revoking token"""
self._token_service.revoke_token(token)
# ============================================================================
# ADAPTERS (Implementation of Ports)
# ============================================================================
# Secondary Adapter - In-memory user storage
class InMemoryUserRepository(UserRepository):
"""Concrete implementation using in-memory storage"""
def __init__(self):
self._users: dict[str, User] = {}
def find_by_username(self, username: str) -> Optional[User]:
"""Find user by username in memory"""
for user in self._users.values():
if user.username == username:
return user
return None
def save(self, user: User) -> User:
"""Save user to memory"""
self._users[user.id] = user
print(f"User {user.username} saved to database")
return user
# Secondary Adapter - Simple token service
class SimpleTokenService(TokenService):
"""Concrete implementation of token service"""
def __init__(self):
self._tokens: dict[str, AuthToken] = {}
def generate_token(self, user_id: str) -> AuthToken:
"""Generate simple token"""
# Create simple token (in production, use JWT or similar)
token_data = f"{user_id}_{datetime.now().timestamp()}"
token = hashlib.md5(token_data.encode()).hexdigest()
auth_token = AuthToken(
token=token,
user_id=user_id,
expires_at=datetime.now() + timedelta(hours=24),
is_valid=True
)
self._tokens[token] = auth_token
return auth_token
def validate_token(self, token: str) -> Optional[AuthToken]:
"""Validate token"""
auth_token = self._tokens.get(token)
if not auth_token:
return None
# Check expiration
if datetime.now() > auth_token.expires_at:
auth_token.is_valid = False
return auth_token
def revoke_token(self, token: str) -> None:
"""Revoke token"""
if token in self._tokens:
self._tokens[token].is_valid = False
print(f"Token {token[:8]}... revoked")
# Secondary Adapter - Simple password service
class SimplePasswordService(PasswordService):
"""Concrete implementation of password service"""
def hash_password(self, password: str) -> str:
"""Hash password using SHA-256 (use bcrypt in production)"""
return hashlib.sha256(password.encode()).hexdigest()
def verify_password(self, password: str, password_hash: str) -> bool:
"""Verify password against hash"""
return self.hash_password(password) == password_hash
# Primary Adapter - Web API controller
class AuthController:
"""Primary adapter handling HTTP requests"""
def __init__(self, auth_service: AuthService):
self._auth_service = auth_service
def register(self, request_data: dict) -> dict:
"""Handle POST /register request"""
try:
user = self._auth_service.register_user(
username=request_data["username"],
email=request_data["email"],
password=request_data["password"]
)
return {
"success": True,
"message": "User registered successfully",
"user_id": user.id
}
except AuthenticationError as e:
return {"success": False, "error": str(e)}
except KeyError as e:
return {"success": False, "error": f"Missing field: {e}"}
def login(self, request_data: dict) -> dict:
"""Handle POST /login request"""
try:
login_request = LoginRequest(
username=request_data["username"],
password=request_data["password"]
)
token = self._auth_service.authenticate(login_request)
return {
"success": True,
"token": token.token,
"expires_at": token.expires_at.isoformat()
}
except AuthenticationError as e:
return {"success": False, "error": str(e)}
except KeyError as e:
return {"success": False, "error": f"Missing field: {e}"}
def validate_session(self, token: str) -> dict:
"""Handle GET /validate request"""
user = self._auth_service.validate_session(token)
if user:
return {
"valid": True,
"user": {
"id": user.id,
"username": user.username,
"email": user.email
}
}
return {"valid": False, "error": "Invalid or expired token"}
def logout(self, token: str) -> dict:
"""Handle POST /logout request"""
self._auth_service.logout(token)
return {"success": True, "message": "Logged out successfully"}
# ============================================================================
# COMPOSITION ROOT
# ============================================================================
def create_auth_system() -> AuthController:
"""Create and wire the authentication system"""
# Create secondary adapters
user_repo = InMemoryUserRepository()
token_service = SimpleTokenService()
password_service = SimplePasswordService()
# Create application core
auth_service = AuthService(
user_repo=user_repo,
token_service=token_service,
password_service=password_service
)
# Create primary adapter
auth_controller = AuthController(auth_service)
return auth_controller
# ============================================================================
# USAGE EXAMPLE
# ============================================================================
if __name__ == "__main__":
# Initialize auth system
auth_controller = create_auth_system()
print("=== User Registration ===")
register_result = auth_controller.register({
"username": "john_doe",
"email": "john@example.com",
"password": "secure123"
})
print(f"Registration result: {register_result}")
if register_result["success"]:
print("\n=== User Login ===")
login_result = auth_controller.login({
"username": "john_doe",
"password": "secure123"
})
print(f"Login result: {login_result}")
if login_result["success"]:
token = login_result["token"]
print("\n=== Session Validation ===")
validation_result = auth_controller.validate_session(token)
print(f"Validation result: {validation_result}")
print("\n=== Logout ===")
logout_result = auth_controller.logout(token)
print(f"Logout result: {logout_result}")
print("\n=== Validation After Logout ===")
validation_after_logout = auth_controller.validate_session(token)
print(f"Validation after logout: {validation_after_logout}")
Key Architectural Elements:
Domain Models: User, AuthToken, and LoginRequest represent core authentication concepts without any external dependencies.
Ports (Interfaces): UserRepository, TokenService, and PasswordService define contracts for user storage, token management, and password operations respectively.
Application Core: AuthService contains the core authentication business logic including user registration rules, credential validation, and session management. It enforces business rules like password strength requirements and account status checks.
Adapters: Concrete implementations handle technical details - InMemoryUserRepository for storage, SimpleTokenService for token generation/validation, and SimplePasswordService for password hashing.
Error Handling: Custom AuthenticationError keeps domain-specific errors separate from technical exceptions.
Security Isolation: The core authentication logic is completely independent of storage mechanisms, token formats, or password hashing algorithms. This makes it easy to upgrade security implementations (like switching from SHA-256 to bcrypt) without touching business logic.
The architecture enables easy testing of authentication rules in isolation and allows switching between different storage systems, token formats, or security mechanisms without affecting the core authentication logic.