Hexagonal Architecture (Ports & Adapters) Practical Guide — The Core of Clean Architecture

Introduction
Core Concepts
- 3 Key Components
- Difference from Layered Architecture
Hands-On Implementation with Python
Testing Strategy
- Domain Unit Tests (No External Dependencies)
- Service Tests with Mocks
Conclusion

Introduction

"We need to switch the database from MySQL to PostgreSQL." "We need to support gRPC in addition to the REST API."

If hearing requirements like these means you have to rewrite the entire codebase, there is a problem with your architecture. Hexagonal Architecture completely decouples business logic from external dependencies, making such changes easy.

Core Concepts

3 Key Components

1. Domain (Core)
   - The heart of business logic
   - No external dependencies (pure code)
   - Entity, Value Object, Domain Service

2. Port
   - Interface (contract) between domain and the outside world
   - Input Port: Outside to Domain (Use Case)
   - Output Port: Domain to Outside (Repository interface)

3. Adapter
   - Concrete implementation of a Port
   - Input Adapter: REST Controller, gRPC Handler, CLI
   - Output Adapter: MySQL Repository, Redis Cache, HTTP Client

Difference from Layered Architecture

# Layered Architecture (Traditional)
# Controller → Service → Repository → DB
# Dependency direction: Top → Bottom (coupled to DB)

# Hexagonal Architecture
# Adapter → Port → Domain ← Port ← Adapter
# Dependency direction: Outside → Inside (Domain is the center)

Hands-On Implementation with Python

Project Structure

order-service/
├── domain/                    # Core Domain
│   ├── models/
│   │   ├── order.py          # Entity
│   │   └── order_item.py     # Value Object
│   └── services/
│       └── order_service.py  # Domain Service
├── ports/                     # Ports (Interfaces)
│   ├── input/
│   │   └── order_use_case.py # Input Port
│   └── output/
│       ├── order_repository.py    # Output Port
│       └── payment_gateway.py     # Output Port
├── adapters/                  # Adapters (Implementations)
│   ├── input/
│   │   ├── rest_controller.py     # REST API
│   │   └── grpc_handler.py        # gRPC
│   └── output/
│       ├── postgres_order_repo.py # PostgreSQL implementation
│       ├── redis_order_cache.py   # Redis cache
│       └── stripe_payment.py      # Stripe payment
└── config/
    └── dependency_injection.py    # DI configuration

Domain Model

# domain/models/order.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List
from uuid import UUID, uuid4


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


@dataclass
class OrderItem:
    product_id: str
    product_name: str
    quantity: int
    unit_price: float

    @property
    def subtotal(self) -> float:
        return self.quantity * self.unit_price


@dataclass
class Order:
    """Order Entity — contains business rules"""
    id: UUID = field(default_factory=uuid4)
    customer_id: str = ""
    items: List[OrderItem] = field(default_factory=list)
    status: OrderStatus = OrderStatus.PENDING
    created_at: datetime = field(default_factory=datetime.now)

    @property
    def total_amount(self) -> float:
        return sum(item.subtotal for item in self.items)

    def add_item(self, item: OrderItem) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError("Cannot add items to a confirmed order")
        if item.quantity <= 0:
            raise ValueError("Quantity must be at least 1")
        self.items.append(item)

    def confirm(self) -> None:
        if not self.items:
            raise ValueError("Cannot confirm an order with no items")
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot confirm from '{self.status.value}' status")
        self.status = OrderStatus.CONFIRMED

    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("Cannot cancel a shipped order")
        self.status = OrderStatus.CANCELLED

Port Definitions

# ports/input/order_use_case.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.models.order import Order, OrderItem


class CreateOrderUseCase(ABC):
    @abstractmethod
    def execute(self, customer_id: str, items: list[OrderItem]) -> Order:
        pass


class ConfirmOrderUseCase(ABC):
    @abstractmethod
    def execute(self, order_id: UUID) -> Order:
        pass


class CancelOrderUseCase(ABC):
    @abstractmethod
    def execute(self, order_id: UUID) -> Order:
        pass


# ports/output/order_repository.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.models.order import Order


class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> None:
        pass

    @abstractmethod
    def find_by_id(self, order_id: UUID) -> Order | None:
        pass

    @abstractmethod
    def find_by_customer(self, customer_id: str) -> list[Order]:
        pass


# ports/output/payment_gateway.py
from abc import ABC, abstractmethod
from uuid import UUID


class PaymentGateway(ABC):
    @abstractmethod
    def charge(self, order_id: UUID, amount: float, customer_id: str) -> bool:
        pass

    @abstractmethod
    def refund(self, order_id: UUID) -> bool:
        pass

Domain Service (Use Case Implementation)

# domain/services/order_service.py
from uuid import UUID
from domain.models.order import Order, OrderItem
from ports.input.order_use_case import (
    CreateOrderUseCase, ConfirmOrderUseCase, CancelOrderUseCase
)
from ports.output.order_repository import OrderRepository
from ports.output.payment_gateway import PaymentGateway


class OrderService(CreateOrderUseCase, ConfirmOrderUseCase, CancelOrderUseCase):
    """Order Service — Input Port implementation"""

    def __init__(
        self,
        order_repo: OrderRepository,
        payment_gateway: PaymentGateway
    ):
        # Depends on Output Port (not the concrete implementation!)
        self._order_repo = order_repo
        self._payment_gateway = payment_gateway

    def execute(self, customer_id: str = None, items: list[OrderItem] = None,
                order_id: UUID = None) -> Order:
        # dispatch based on params (simplified)
        if customer_id and items:
            return self._create_order(customer_id, items)
        raise ValueError("Invalid parameters")

    def _create_order(self, customer_id: str, items: list[OrderItem]) -> Order:
        order = Order(customer_id=customer_id)
        for item in items:
            order.add_item(item)
        self._order_repo.save(order)
        return order

    def confirm_order(self, order_id: UUID) -> Order:
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order not found: {order_id}")

        # Process payment
        success = self._payment_gateway.charge(
            order_id=order.id,
            amount=order.total_amount,
            customer_id=order.customer_id
        )
        if not success:
            raise ValueError("Payment failed")

        order.confirm()
        self._order_repo.save(order)
        return order

    def cancel_order(self, order_id: UUID) -> Order:
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order not found: {order_id}")

        order.cancel()
        self._payment_gateway.refund(order_id)
        self._order_repo.save(order)
        return order

Adapter Implementation

# adapters/output/postgres_order_repo.py
import psycopg2
from uuid import UUID
from domain.models.order import Order, OrderItem, OrderStatus
from ports.output.order_repository import OrderRepository


class PostgresOrderRepository(OrderRepository):
    def __init__(self, connection_string: str):
        self._conn_str = connection_string

    def save(self, order: Order) -> None:
        with psycopg2.connect(self._conn_str) as conn:
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO orders (id, customer_id, status, created_at)
                    VALUES (%s, %s, %s, %s)
                    ON CONFLICT (id) DO UPDATE SET status = %s
                """, (str(order.id), order.customer_id,
                      order.status.value, order.created_at,
                      order.status.value))

                for item in order.items:
                    cur.execute("""
                        INSERT INTO order_items
                        (order_id, product_id, product_name, quantity, unit_price)
                        VALUES (%s, %s, %s, %s, %s)
                        ON CONFLICT DO NOTHING
                    """, (str(order.id), item.product_id,
                          item.product_name, item.quantity, item.unit_price))

    def find_by_id(self, order_id: UUID) -> Order | None:
        with psycopg2.connect(self._conn_str) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT * FROM orders WHERE id = %s", (str(order_id),))
                row = cur.fetchone()
                if not row:
                    return None
                return self._to_domain(row, cur)

    def find_by_customer(self, customer_id: str) -> list[Order]:
        # Implementation omitted
        pass

    def _to_domain(self, row, cursor) -> Order:
        # Convert DB row to domain model
        pass


# adapters/input/rest_controller.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import UUID
from domain.services.order_service import OrderService
from domain.models.order import OrderItem

app = FastAPI()


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


class OrderResponse(BaseModel):
    id: str
    customer_id: str
    status: str
    total_amount: float


def create_rest_controller(order_service: OrderService):

    @app.post("/orders", response_model=OrderResponse)
    async def create_order(request: CreateOrderRequest):
        items = [
            OrderItem(
                product_id=i["product_id"],
                product_name=i["product_name"],
                quantity=i["quantity"],
                unit_price=i["unit_price"]
            )
            for i in request.items
        ]
        order = order_service._create_order(request.customer_id, items)
        return OrderResponse(
            id=str(order.id),
            customer_id=order.customer_id,
            status=order.status.value,
            total_amount=order.total_amount
        )

    @app.post("/orders/{order_id}/confirm")
    async def confirm_order(order_id: UUID):
        try:
            order = order_service.confirm_order(order_id)
            return {"status": order.status.value}
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))

    return app

Dependency Injection Configuration

# config/dependency_injection.py
from domain.services.order_service import OrderService
from adapters.output.postgres_order_repo import PostgresOrderRepository
from adapters.output.stripe_payment import StripePaymentGateway
from adapters.input.rest_controller import create_rest_controller


def bootstrap():
    # Output Adapters
    order_repo = PostgresOrderRepository(
        connection_string="postgresql://user:pass@localhost/orders"
    )
    payment_gateway = StripePaymentGateway(
        api_key="sk_test_xxx"
    )

    # Domain Service (Port implementation)
    order_service = OrderService(
        order_repo=order_repo,
        payment_gateway=payment_gateway
    )

    # Input Adapter
    app = create_rest_controller(order_service)

    return app


# Want to switch databases?
# Just replace PostgresOrderRepository with MongoOrderRepository!
# No changes to domain code!

Testing Strategy

Domain Unit Tests (No External Dependencies)

import pytest
from domain.models.order import Order, OrderItem, OrderStatus


class TestOrder:
    def test_add_item(self):
        order = Order(customer_id="C001")
        item = OrderItem("P001", "Laptop", 1, 1500000)
        order.add_item(item)
        assert len(order.items) == 1
        assert order.total_amount == 1500000

    def test_cannot_add_item_to_confirmed_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "Laptop", 1, 1500000))
        order.confirm()
        with pytest.raises(ValueError, match="confirmed order"):
            order.add_item(OrderItem("P002", "Mouse", 1, 50000))

    def test_cannot_confirm_empty_order(self):
        order = Order(customer_id="C001")
        with pytest.raises(ValueError, match="no items"):
            order.confirm()

    def test_cannot_cancel_shipped_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "Laptop", 1, 1500000))
        order.confirm()
        order.status = OrderStatus.SHIPPED
        with pytest.raises(ValueError, match="shipped"):
            order.cancel()

Service Tests with Mocks

from unittest.mock import MagicMock
from domain.services.order_service import OrderService
from domain.models.order import Order, OrderItem


class TestOrderService:
    def setup_method(self):
        self.mock_repo = MagicMock()
        self.mock_payment = MagicMock()
        self.service = OrderService(self.mock_repo, self.mock_payment)

    def test_create_order(self):
        items = [OrderItem("P001", "Laptop", 1, 1500000)]
        order = self.service._create_order("C001", items)

        assert order.customer_id == "C001"
        assert len(order.items) == 1
        self.mock_repo.save.assert_called_once()

    def test_confirm_order_with_payment(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "Laptop", 1, 1500000))
        self.mock_repo.find_by_id.return_value = order
        self.mock_payment.charge.return_value = True

        result = self.service.confirm_order(order.id)

        self.mock_payment.charge.assert_called_once()
        assert result.status.value == "confirmed"

Conclusion

The core values of Hexagonal Architecture:

Domain Independence: Business logic is not coupled to DB or frameworks
Easy Replacement: Just swap adapters to change external systems
Testability: Domain tests need no mocks; service tests only mock ports
Ports as Contracts: Interfaces (Ports) serve as clear contracts between internal and external boundaries

Quiz (6 Questions)

Q1. What are the three key components of Hexagonal Architecture? Domain (Core), Port (Interface), Adapter (Implementation)

Q2. What is the difference between Input Port and Output Port? Input Port: From outside to domain (Use Case). Output Port: From domain to outside (Repository interface)

Q3. What is the direction of dependencies in Hexagonal Architecture? Outside (Adapter) to inside (Domain). The domain does not depend on anything external.

Q4. What needs to change when switching the DB from MySQL to PostgreSQL? Only the Output Adapter needs to be replaced (no changes to domain code)

Q5. Why are mocks unnecessary in domain unit tests? Because domain models have no external dependencies, so only pure logic is tested

Q6. What is the main advantage of Hexagonal over Layered Architecture? Business logic is not coupled to DB/frameworks, making replacement and testing easier