Hexagonal Architecture (Ports & Adapters)

What is Hexagonal Architecture?

The Simple Explanation (ELI5)

Imagine your video game console:

• The console (core): Runs your games with all the game logic
• Ports: HDMI port, controller ports, power port
• Adapters: HDMI cable, wireless controller, power adapter

You can swap adapters (use different TVs, controllers) without changing the console. That’s hexagonal architecture - your business logic is the console, everything else can be swapped!

The Definition

Hexagonal Architecture (also called Ports and Adapters) is an architectural pattern where:

1. Business logic is at the center (the “hexagon”)
2. Ports define interfaces for communication
3. Adapters implement those interfaces for specific technologies
4. Dependencies point inward toward business logic

Created by Alistair Cockburn in 2005. The “hexagon” shape has no special meaning - it just needs enough sides for drawing ports!

---

The Hexagonal Structure

---

Ports and Adapters Explained

Ports (Interfaces)

A port is an interface that defines HOW to interact with the application core.

Two Types:

1. Primary Ports (Driving Ports)

   • Purpose: What the application offers to the outside world
   • Direction: External → Core (adapters CALL these)
   • Examples: Use case interfaces, command handlers

2. Secondary Ports (Driven Ports)

   • Purpose: What the application needs from the outside world
   • Direction: Core → External (adapters IMPLEMENT these)
   • Examples: Repository interfaces, notification interfaces

Adapters (Implementations)

An adapter is a concrete implementation of a port for a specific technology.

Two Types:

1. Primary Adapters (Driving Adapters)

   • Convert external requests → port calls
   • Examples: REST controller, GraphQL resolver, CLI command

2. Secondary Adapters (Driven Adapters)

   • Implement ports using specific technology
   • Examples: PostgreSQL repository, SendGrid email sender

---

Hexagonal Architecture in Code

Example: Order Management System

1. Application Core (The Hexagon)

Python

core/domain/order.py

from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from typing import List

class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"

@dataclass
class OrderItem:
    product_id: str
    quantity: int
    price: Decimal

class Order:
    """Domain entity - pure business logic"""

    def __init__(self, id: str, customer_id: str):
        self.id = id
        self.customer_id = customer_id
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING

    def add_item(self, product_id: str, quantity: int, price: Decimal):
        """Business rule: Cannot add items to confirmed order"""
        if self.status == OrderStatus.CONFIRMED:
            raise ValueError("Cannot modify confirmed order")

        self.items.append(OrderItem(product_id, quantity, price))

    def calculate_total(self) -> Decimal:
        return sum(item.price * item.quantity for item in self.items)

    def confirm(self):
        """Business rule: Must have items to confirm"""
        if not self.items:
            raise ValueError("Cannot confirm empty order")
        if self.status != OrderStatus.PENDING:
            raise ValueError("Order already confirmed")

        self.status = OrderStatus.CONFIRMED

Java

core/domain/Order.java

public class Order {
    /**
     * Domain entity - pure business logic
     */

    private final String id;
    private final String customerId;
    private final List<OrderItem> items;
    private OrderStatus status;

    public Order(String id, String customerId) {
        this.id = id;
        this.customerId = customerId;
        this.items = new ArrayList<>();
        this.status = OrderStatus.PENDING;
    }

    public void addItem(String productId, int quantity, BigDecimal price) {
        // Business rule: Cannot add items to confirmed order
        if (status == OrderStatus.CONFIRMED) {
            throw new IllegalStateException("Cannot modify confirmed order");
        }

        items.add(new OrderItem(productId, quantity, price));
    }

    public BigDecimal calculateTotal() {
        return items.stream()
            .map(item -> item.price().multiply(BigDecimal.valueOf(item.quantity())))
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }

    public void confirm() {
        // Business rule: Must have items to confirm
        if (items.isEmpty()) {
            throw new IllegalStateException("Cannot confirm empty order");
        }
        if (status != OrderStatus.PENDING) {
            throw new IllegalStateException("Order already confirmed");
        }

        status = OrderStatus.CONFIRMED;
    }
}

2. Primary Port (What App Offers)

Python

core/ports/primary/order_use_cases.py

from abc import ABC, abstractmethod
from typing import List

class OrderUseCases(ABC):
    """Primary port - defines what the application offers"""

    @abstractmethod
    async def place_order(
        self,
        customer_id: str,
        items: List[dict]
    ) -> Order:
        """Place a new order"""
        pass

    @abstractmethod
    async def confirm_order(self, order_id: str) -> Order:
        """Confirm an order"""
        pass

    @abstractmethod
    async def cancel_order(self, order_id: str, reason: str) -> Order:
        """Cancel an order"""
        pass

    @abstractmethod
    async def get_order(self, order_id: str) -> Order:
        """Retrieve order by ID"""
        pass

Java

core/ports/primary/OrderUseCases.java

public interface OrderUseCases {
    /**
     * Primary port - defines what the application offers
     */

    /**
     * Place a new order
     */
    Order placeOrder(String customerId, List<OrderItemRequest> items);

    /**
     * Confirm an order
     */
    Order confirmOrder(String orderId);

    /**
     * Cancel an order
     */
    Order cancelOrder(String orderId, String reason);

    /**
     * Retrieve order by ID
     */
    Optional<Order> getOrder(String orderId);
}

3. Secondary Ports (What App Needs)

Python

core/ports/secondary/order_repository.py

from abc import ABC, abstractmethod
from typing import Optional

class OrderRepository(ABC):
    """Secondary port - defines what the application needs for persistence"""

    @abstractmethod
    async def save(self, order: Order) -> None:
        """Persist an order"""
        pass

    @abstractmethod
    async def find_by_id(self, order_id: str) -> Optional[Order]:
        """Find order by ID"""
        pass

class NotificationService(ABC):
    """Secondary port - defines what the application needs for notifications"""

    @abstractmethod
    async def notify_order_confirmed(self, order: Order) -> None:
        """Send confirmation notification"""
        pass

    @abstractmethod
    async def notify_order_cancelled(self, order: Order, reason: str) -> None:
        """Send cancellation notification"""
        pass

class PaymentService(ABC):
    """Secondary port - defines what the application needs for payments"""

    @abstractmethod
    async def process_payment(self, order_id: str, amount: Decimal) -> bool:
        """Process payment for order"""
        pass

Java

core/ports/secondary/OrderRepository.java

public interface OrderRepository {
    /**
     * Secondary port - defines what the application needs for persistence
     */
    void save(Order order);
    Optional<Order> findById(String orderId);
}

public interface NotificationService {
    /**
     * Secondary port - defines what the application needs for notifications
     */
    void notifyOrderConfirmed(Order order);
    void notifyOrderCancelled(Order order, String reason);
}

public interface PaymentService {
    /**
     * Secondary port - defines what the application needs for payments
     */
    boolean processPayment(String orderId, BigDecimal amount);
}

4. Use Case Implementation (Application Core)

Python

core/application/order_service.py

class OrderService(OrderUseCases):
    """
    Implements primary port (what app offers)
    Uses secondary ports (what app needs)
    """

    def __init__(
        self,
        order_repo: OrderRepository,
        payment_service: PaymentService,
        notification_service: NotificationService
    ):
        # Depend on INTERFACES (ports), not implementations!
        self._order_repo = order_repo
        self._payment = payment_service
        self._notifications = notification_service

    async def place_order(
        self,
        customer_id: str,
        items: List[dict]
    ) -> Order:
        """Use case implementation"""

        # Create order (domain logic)
        order = Order(id=generate_id(), customer_id=customer_id)

        for item in items:
            order.add_item(
                product_id=item['product_id'],
                quantity=item['quantity'],
                price=Decimal(item['price'])
            )

        # Save order (use secondary port)
        await self._order_repo.save(order)

        return order

    async def confirm_order(self, order_id: str) -> Order:
        """Use case implementation"""

        # Load order (use secondary port)
        order = await self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")

        # Confirm order (domain logic)
        order.confirm()

        # Process payment (use secondary port)
        payment_success = await self._payment.process_payment(
            order.id,
            order.calculate_total()
        )

        if not payment_success:
            raise ValueError("Payment failed")

        # Save confirmed order (use secondary port)
        await self._order_repo.save(order)

        # Send notification (use secondary port)
        await self._notifications.notify_order_confirmed(order)

        return order

    async def cancel_order(self, order_id: str, reason: str) -> Order:
        """Use case implementation"""

        order = await self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")

        order.cancel()
        await self._order_repo.save(order)
        await self._notifications.notify_order_cancelled(order, reason)

        return order

    async def get_order(self, order_id: str) -> Order:
        """Use case implementation"""
        order = await self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")
        return order

Java

core/application/OrderService.java

public class OrderService implements OrderUseCases {
    /**
     * Implements primary port (what app offers)
     * Uses secondary ports (what app needs)
     */

    private final OrderRepository orderRepo;
    private final PaymentService payment;
    private final NotificationService notifications;

    public OrderService(
        OrderRepository orderRepo,
        PaymentService payment,
        NotificationService notifications
    ) {
        // Depend on INTERFACES (ports), not implementations!
        this.orderRepo = orderRepo;
        this.payment = payment;
        this.notifications = notifications;
    }

    @Override
    public Order placeOrder(String customerId, List<OrderItemRequest> items) {
        // Create order (domain logic)
        Order order = new Order(generateId(), customerId);

        for (OrderItemRequest item : items) {
            order.addItem(
                item.productId(),
                item.quantity(),
                item.price()
            );
        }

        // Save order (use secondary port)
        orderRepo.save(order);

        return order;
    }

    @Override
    public Order confirmOrder(String orderId) {
        // Load order (use secondary port)
        Order order = orderRepo.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));

        // Confirm order (domain logic)
        order.confirm();

        // Process payment (use secondary port)
        boolean paymentSuccess = payment.processPayment(
            order.getId(),
            order.calculateTotal()
        );

        if (!paymentSuccess) {
            throw new PaymentFailedException("Payment failed");
        }

        // Save confirmed order (use secondary port)
        orderRepo.save(order);

        // Send notification (use secondary port)
        notifications.notifyOrderConfirmed(order);

        return order;
    }

    @Override
    public Order cancelOrder(String orderId, String reason) {
        Order order = orderRepo.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));

        order.cancel();
        orderRepo.save(order);
        notifications.notifyOrderCancelled(order, reason);

        return order;
    }

    @Override
    public Optional<Order> getOrder(String orderId) {
        return orderRepo.findById(orderId);
    }
}

5. Primary Adapter (REST API)

Python

adapters/primary/rest_api.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class CreateOrderRequest(BaseModel):
    customer_id: str
    items: list[dict]

class OrderResponse(BaseModel):
    order_id: str
    status: str
    total: float

@app.post("/orders", response_model=OrderResponse)
async def create_order(request: CreateOrderRequest):
    """Primary adapter - converts HTTP → port call"""
    try:
        # Call primary port
        order = await order_use_cases.place_order(
            customer_id=request.customer_id,
            items=request.items
        )

        # Convert domain model → DTO
        return OrderResponse(
            order_id=order.id,
            status=order.status.value,
            total=float(order.calculate_total())
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/orders/{order_id}/confirm")
async def confirm_order(order_id: str):
    """Primary adapter - converts HTTP → port call"""
    try:
        order = await order_use_cases.confirm_order(order_id)
        return OrderResponse(
            order_id=order.id,
            status=order.status.value,
            total=float(order.calculate_total())
        )
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

Java

adapters/primary/RestApiAdapter.java

@RestController
@RequestMapping("/orders")
public class RestApiAdapter {
    /**
     * Primary adapter - converts HTTP → port call
     */

    private final OrderUseCases orderUseCases;

    public record CreateOrderRequest(
        String customerId,
        List<OrderItemRequest> items
    ) {}

    public record OrderResponse(
        String orderId,
        String status,
        BigDecimal total
    ) {}

    @PostMapping
    public ResponseEntity<OrderResponse> createOrder(
        @RequestBody CreateOrderRequest request
    ) {
        try {
            // Call primary port
            Order order = orderUseCases.placeOrder(
                request.customerId(),
                request.items()
            );

            // Convert domain model → DTO
            OrderResponse response = new OrderResponse(
                order.getId(),
                order.getStatus().toString(),
                order.calculateTotal()
            );

            return ResponseEntity.ok(response);
        } catch (IllegalArgumentException e) {
            return ResponseEntity.badRequest().build();
        }
    }

    @PostMapping("/{orderId}/confirm")
    public ResponseEntity<OrderResponse> confirmOrder(
        @PathVariable String orderId
    ) {
        try {
            Order order = orderUseCases.confirmOrder(orderId);
            OrderResponse response = new OrderResponse(
                order.getId(),
                order.getStatus().toString(),
                order.calculateTotal()
            );
            return ResponseEntity.ok(response);
        } catch (OrderNotFoundException e) {
            return ResponseEntity.notFound().build();
        }
    }
}

6. Secondary Adapter (PostgreSQL Repository)

Python

adapters/secondary/postgres_order_repository.py

from sqlalchemy.orm import Session
from typing import Optional

class PostgresOrderRepository(OrderRepository):
    """Secondary adapter - implements port using PostgreSQL"""

    def __init__(self, db_session: Session):
        self._session = db_session

    async def save(self, order: Order) -> None:
        """Convert domain → database model"""
        db_order = OrderModel(
            id=order.id,
            customer_id=order.customer_id,
            status=order.status.value
        )

        # Save order items
        for item in order.items:
            db_item = OrderItemModel(
                order_id=order.id,
                product_id=item.product_id,
                quantity=item.quantity,
                price=item.price
            )
            db_order.items.append(db_item)

        self._session.add(db_order)
        await self._session.commit()

    async def find_by_id(self, order_id: str) -> Optional[Order]:
        """Convert database model → domain"""
        db_order = await self._session.query(OrderModel)\
            .filter_by(id=order_id)\
            .first()

        if not db_order:
            return None

        # Reconstruct domain entity
        order = Order(
            id=db_order.id,
            customer_id=db_order.customer_id
        )
        order.status = OrderStatus(db_order.status)

        for db_item in db_order.items:
            order.items.append(OrderItem(
                product_id=db_item.product_id,
                quantity=db_item.quantity,
                price=db_item.price
            ))

        return order

Java

adapters/secondary/JpaOrderRepository.java

@Repository
public class JpaOrderRepository implements OrderRepository {
    /**
     * Secondary adapter - implements port using JPA/PostgreSQL
     */

    @PersistenceContext
    private EntityManager entityManager;

    @Override
    public void save(Order order) {
        // Convert domain → JPA entity
        OrderEntity entity = new OrderEntity();
        entity.setId(order.getId());
        entity.setCustomerId(order.getCustomerId());
        entity.setStatus(order.getStatus().toString());

        // Save order items
        for (OrderItem item : order.getItems()) {
            OrderItemEntity itemEntity = new OrderItemEntity();
            itemEntity.setOrderId(order.getId());
            itemEntity.setProductId(item.productId());
            itemEntity.setQuantity(item.quantity());
            itemEntity.setPrice(item.price());
            entity.getItems().add(itemEntity);
        }

        entityManager.persist(entity);
        entityManager.flush();
    }

    @Override
    public Optional<Order> findById(String orderId) {
        // Query database
        OrderEntity entity = entityManager.find(OrderEntity.class, orderId);

        if (entity == null) {
            return Optional.empty();
        }

        // Reconstruct domain entity
        Order order = new Order(
            entity.getId(),
            entity.getCustomerId()
        );
        order.setStatus(OrderStatus.valueOf(entity.getStatus()));

        for (OrderItemEntity itemEntity : entity.getItems()) {
            order.getItems().add(new OrderItem(
                itemEntity.getProductId(),
                itemEntity.getQuantity(),
                itemEntity.getPrice()
            ));
        }

        return Optional.of(order);
    }
}

7. Secondary Adapter (Email Notification)

Python

adapters/secondary/email_notification_service.py

import smtplib
from email.message import EmailMessage

class EmailNotificationService(NotificationService):
    """Secondary adapter - implements port using SMTP"""

    def __init__(self, smtp_host: str, smtp_port: int):
        self._host = smtp_host
        self._port = smtp_port

    async def notify_order_confirmed(self, order: Order) -> None:
        """Send confirmation email"""
        message = EmailMessage()
        message['Subject'] = f'Order {order.id} Confirmed'
        message['To'] = await self._get_customer_email(order.customer_id)
        message.set_content(
            f'Your order {order.id} has been confirmed.\n'
            f'Total: ${order.calculate_total()}'
        )

        with smtplib.SMTP(self._host, self._port) as smtp:
            smtp.send_message(message)

    async def notify_order_cancelled(
        self,
        order: Order,
        reason: str
    ) -> None:
        """Send cancellation email"""
        message = EmailMessage()
        message['Subject'] = f'Order {order.id} Cancelled'
        message['To'] = await self._get_customer_email(order.customer_id)
        message.set_content(
            f'Your order {order.id} has been cancelled.\n'
            f'Reason: {reason}'
        )

        with smtplib.SMTP(self._host, self._port) as smtp:
            smtp.send_message(message)

Java

adapters/secondary/EmailNotificationService.java

@Service
public class EmailNotificationService implements NotificationService {
    /**
     * Secondary adapter - implements port using JavaMail
     */

    private final JavaMailSender mailSender;

    @Override
    public void notifyOrderConfirmed(Order order) {
        SimpleMailMessage message = new SimpleMailMessage();
        message.setTo(getCustomerEmail(order.getCustomerId()));
        message.setSubject("Order " + order.getId() + " Confirmed");
        message.setText(
            String.format(
                "Your order %s has been confirmed.%nTotal: $%s",
                order.getId(),
                order.calculateTotal()
            )
        );

        mailSender.send(message);
    }

    @Override
    public void notifyOrderCancelled(Order order, String reason) {
        SimpleMailMessage message = new SimpleMailMessage();
        message.setTo(getCustomerEmail(order.getCustomerId()));
        message.setSubject("Order " + order.getId() + " Cancelled");
        message.setText(
            String.format(
                "Your order %s has been cancelled.%nReason: %s",
                order.getId(),
                reason
            )
        );

        mailSender.send(message);
    }

    private String getCustomerEmail(String customerId) {
        // Fetch from customer service
        return customerService.getCustomer(customerId).getEmail();
    }
}

---

Advantages of Hexagonal Architecture

Technology Independence

Swap databases, change UI framework, switch messaging systems - all without touching business logic!

Testability

Test business logic in isolation. Mock all adapters. No need for database or HTTP server in tests.

Flexibility

Add new adapters easily. Support REST and GraphQL simultaneously. Use multiple databases.

Clear Boundaries

Business logic is completely isolated. No accidental dependencies on infrastructure.

---

Hexagonal vs Layered Architecture

Aspect	Layered Architecture	Hexagonal Architecture
Structure	Horizontal layers	Center + adapters
Dependencies	Top-down through layers	All point inward to core
Flexibility	Less flexible	More flexible
Infrastructure	Bottom layer	Adapters on outside
Testing	Test through layers	Test core directly
Swap Components	Harder	Easier
Complexity	Simpler	More complex
Best For	Traditional apps	Event-driven, multiple interfaces

---

Key Takeaways

Business Logic First

Core is pure business logic with zero infrastructure dependencies. Everything else is just an implementation detail!

Ports Define Contracts

Interfaces (ports) define HOW to interact with the core. Adapters provide specific implementations.

Easily Swappable

Change databases, UI frameworks, or messaging systems without touching business logic. True technology independence!

Testing Paradise

Test business logic without database, HTTP, or any external dependencies. Fast, reliable, isolated tests.

---

Related Topics

• Layered Architecture - Traditional alternative
• Domain-Driven Design - Perfect fit for hexagonal
• Dependency Inversion - Core principle
• Adapter Pattern - The pattern behind adapters

---

Further Reading

• “Hexagonal Architecture” by Alistair Cockburn (original article)
• “Get Your Hands Dirty on Clean Architecture” by Tom Hombergs
• “Clean Architecture” by Robert C. Martin
• “Implementing Domain-Driven Design” by Vaughn Vernon