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

• Introduction
• Core Concepts
  • 3 Key Components
  • Difference from Layered Architecture
• The Dependency Direction Is the Whole Idea
  • There Are Two Arrows, Not One
  • What Breaks Without This Rule
• Driving Ports and Driven Ports
• Hands-On Implementation with Python
  • Project Structure
  • Domain Model
  • Port Definitions
  • Domain Service (Use Case Implementation)
  • Adapter Implementation
  • Dependency Injection Configuration
• Testing Strategy
  • Domain Unit Tests (No External Dependencies)
  • Service Tests with Mocks
• Following One Requirement End to End
• Pitfalls and Failure Modes
  • Pitfall 1. The Anemic Domain
  • Pitfall 2. The Leaky Port
  • Pitfall 3. DTO Explosion
  • Pitfall 4. Transaction Boundaries
• When Not to Use This
• Conclusion
• References
• Quiz

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)

The Dependency Direction Is the Whole Idea

Because of the name "ports and adapters", it is easy to mistake this for "the pattern where you write a lot of interfaces". But the only rule this architecture actually enforces is the direction of dependencies. Alistair Cockburn's original write-up states the intent in a single sentence: the application must be equally drivable by users, programs, automated tests, or batch scripts, and must be developable and testable in isolation from the run-time devices and databases it will eventually be attached to.

There Are Two Arrows, Not One

Most of the confusion comes from thinking there is only one arrow. There are actually two, and they point in opposite directions. One is the import arrow — which module imports which. The other is the call arrow — who invokes whose methods at run time. The import arrow always points inward; the call arrow points outward.

[Import direction]   adapters ──import──▶ ports ──import──▶ domain
                     (outer)                                 (inner)

[Call direction]     domain ──call──▶ ports (abstract types) ──▶ adapters
                     (inner)                                     (outer)

The same boundary is crossed by two arrows going opposite ways.
That opposition is exactly what dependency inversion means.

Check it against the code above. The abstract class OrderRepository lives under ports/output/ — it is domain-side code. The implementation PostgresOrderRepository lives under adapters/output/ and imports OrderRepository, which sits further inward than itself. The import arrow came from the outside in. At run time, however, OrderService calls save() and that call actually executes inside the PostgreSQL adapter. The call arrow goes from the inside out. Two arrows pointing opposite ways across the same boundary — that is dependency inversion.

This is also why we call a port "an interface the domain owns". A port is not an API that infrastructure offers to the domain; it is a contract the domain demands of infrastructure. That is why a port's name and signature must be written in the vocabulary of the domain. find_by_customer(customer_id) speaks the domain's language; execute_query(sql) speaks infrastructure's. Once infrastructure vocabulary creeps into a port, what you have is a port in name only.

What Breaks Without This Rule

Drop dependency inversion and domain code usually ends up looking like this.

# domain/models/order.py  ← the domain, yet it imports infrastructure
from sqlalchemy.orm import Session           # ORM session type
from infrastructure.db import OrderTable     # ORM mapping class


class Order:
    def confirm(self, session: Session) -> None:
        if not self.items:
            raise ValueError("Cannot confirm an order with no items")
        self.status = "confirmed"
        session.merge(OrderTable.from_domain(self))   # the domain knows about commits
        session.commit()

Two concrete things break here.

First, you can no longer unit-test the domain. The moment you import that module the ORM loads, and an ORM usually demands a configured connection or metadata. To verify a one-line rule like "an order with no items cannot be confirmed", you end up standing up a test database. Tests get slow, and slow tests eventually stop being run at all.

Second, a library upgrade becomes a domain change. When the ORM's major version bumps and the session API changes, domain files show up in the diff. A reviewer cannot tell whether a business rule changed or plumbing changed. After a few such diffs, the git history of your domain files is a history of library migrations rather than a history of rules.

So the familiar marketing line — "it makes swapping the database easy" — is, strictly speaking, a side effect. Most teams finish a service without ever moving from PostgreSQL to MongoDB. The real payoff is that you can read, test, and change domain rules without infrastructure in the room.

Driving Ports and Driven Ports

Cockburn splits ports into two kinds: primary (driving) and secondary (driven). The definitions are compact: a primary actor is the one that drives the application, and a secondary actor is the one the application drives.

In practice, one test is enough to tell them apart: who starts the interaction?

Driving (primary) ports — the outside starts the call
    REST controller / gRPC handler / CLI / batch job / queue consumer / acceptance test
              │
              ▼
      ┌────────────────────────────┐
      │        Application         │
      │   (use cases + domain)     │
      └────────────────────────────┘
              │
              ▼
Driven (secondary) ports — the application starts the call
    OrderRepository / PaymentGateway / NotificationSender / Clock

The test: who opened the conversation?

Real examples on the driving side look like this. A REST controller receiving an HTTP request is started by the user. A consumer pulling messages off a queue looks as though the broker is pushing into it, but it is the side that starts an application use case, so it is a driving port. A cron-driven settlement batch and an operator's CLI command are driving too. So is an acceptance test — a point the original write-up emphasizes in particular. The test harness plugs into exactly the same socket as the REST controller.

The driven side looks like this. OrderRepository stores and retrieves orders. PaymentGateway authorizes and refunds payments. NotificationSender sends email or push. The one people routinely forget to include here is the clock, together with the random number generator. If the domain reads the current time directly, its tests see a different value on every run. Move time and randomness behind driven ports and the tests become deterministic.

There are ambiguous cases. For a single message queue, the consumer side is a driving port while the producer side is a driven port. Using the same technology does not make it the same kind of port. What decides the direction is not the technology but where the call originates.

On how many ports to have, the original is deliberately cautious: it says there does not appear to be any particular damage in picking the "wrong" number, so it remains a matter of intuition, and that the author himself prefers two or three, at most four. There is no rule anywhere in the original saying you must mechanically create one port per external system. Ports are divided by kinds of conversation, not by counts of adapters.

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"

Following One Requirement End to End

Let us thread every piece so far through a single requirement: "cancelling an order refunds the payment and puts the stock back."

Start by deciding which file each piece lands in. The project structure from earlier is used as is.

Requirement: cancelling an order refunds the payment and restocks inventory

adapters/input/rest_controller.py       receives DELETE /orders/{id}   ← driving adapter
ports/input/order_use_case.py           CancelOrderUseCase contract    ← driving port
domain/services/order_service.py        orchestrates the cancellation  ← use case
domain/models/order.py                  Order.cancel() decides rules   ← domain
ports/output/order_repository.py        save / find_by_id              ← driven port
ports/output/payment_gateway.py         refund                         ← driven port
ports/output/inventory_port.py          restock  (newly added here)    ← driven port
adapters/output/postgres_order_repo.py  the actual SQL                 ← driven adapter
adapters/output/stripe_payment.py       the actual Stripe call         ← driven adapter
adapters/output/wms_inventory.py        the actual WMS HTTP call       ← driven adapter

All the new requirement added is one driven port and one adapter for it. A requirement to restock inventory does not change the cancellation rule inside the domain model. Whether that separation holds is the first signal that the design is still alive.

# ports/output/inventory_port.py — a contract the domain demands of infrastructure
from abc import ABC, abstractmethod
from uuid import UUID


class InventoryPort(ABC):
    @abstractmethod
    def restock(self, order_id: UUID, lines: list[tuple[str, int]]) -> None:
        """Put back stock for each (product_id, quantity) pair."""


# domain/models/order.py — rules are decided here and only here
    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("Cannot cancel a shipped order")
        if self.status == OrderStatus.CANCELLED:
            raise ValueError("Order is already cancelled")
        self.status = OrderStatus.CANCELLED

    def restock_lines(self) -> list[tuple[str, int]]:
        return [(i.product_id, i.quantity) for i in self.items]


# domain/services/order_service.py — orchestrates only, never decides rules
class CancelOrderService(CancelOrderUseCase):
    def __init__(self, orders: OrderRepository,
                 payments: PaymentGateway,
                 inventory: InventoryPort):
        self._orders = orders
        self._payments = payments
        self._inventory = inventory

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

        order.cancel()                        # the domain decides
        self._orders.save(order)              # driven port 1
        self._payments.refund(order.id)       # driven port 2
        self._inventory.restock(order.id, order.restock_lines())  # driven port 3
        return order

Notice that the use case contains no conditional at all. The judgement "a shipped order cannot be cancelled" lives inside Order.cancel(), and the use case only sequences steps and calls ports. The collapse of that boundary is the first pitfall in the next section.

Now verify the entire requirement without starting a single piece of infrastructure. Using hand-written fake adapters instead of a mocking library makes what was called visible directly in the assertions, which makes failure messages far easier to read.

# tests/test_cancel_order.py — the whole path, using only fakes
import pytest
from domain.models.order import Order, OrderItem, OrderStatus
from domain.services.order_service import CancelOrderService


class FakeOrderRepository:
    def __init__(self, order):
        self._order, self.saved = order, []

    def find_by_id(self, order_id):
        return self._order

    def save(self, order):
        self.saved.append(order.status)


class FakePaymentGateway:
    def __init__(self):
        self.refunded = []

    def refund(self, order_id):
        self.refunded.append(order_id)
        return True


class FakeInventory:
    def __init__(self):
        self.restocked = []

    def restock(self, order_id, lines):
        self.restocked.append(lines)


def make_order() -> Order:
    order = Order(customer_id="C001")
    order.add_item(OrderItem("P001", "Laptop", 1, 1_500_000))
    order.add_item(OrderItem("P002", "Mouse", 2, 50_000))
    return order


def test_cancel_refunds_and_restocks():
    order = make_order()
    repo, pay, inv = FakeOrderRepository(order), FakePaymentGateway(), FakeInventory()
    service = CancelOrderService(repo, pay, inv)

    result = service.execute(order.id)

    assert result.status is OrderStatus.CANCELLED
    assert repo.saved == [OrderStatus.CANCELLED]
    assert pay.refunded == [order.id]
    assert inv.restocked == [[("P001", 1), ("P002", 2)]]


def test_shipped_order_is_not_cancellable():
    order = make_order()
    order.status = OrderStatus.SHIPPED
    service = CancelOrderService(FakeOrderRepository(order),
                                 FakePaymentGateway(), FakeInventory())

    with pytest.raises(ValueError, match="shipped"):
        service.execute(order.id)

Here is the expected output. The shape below is what pytest 8.x prints; the second run is the same suite after deliberately deleting the refund call from the use case.

$ pytest tests/test_cancel_order.py -q
..                                                            [100%]
2 passed in 0.03s

$ pytest tests/test_cancel_order.py -q      # after deleting the refund call
.F                                                            [100%]
=================================== FAILURES ===================================
_____________________ test_cancel_refunds_and_restocks ________________________
E       AssertionError: assert [] == [UUID('9f0c1a3e-...')]
E         Right contains one more item: UUID('9f0c1a3e-...')
tests/test_cancel_order.py:57: AssertionError
1 failed, 1 passed in 0.04s

The number to look at is the elapsed time. No database, no payment provider, no WMS — it finishes in milliseconds. Only tests this fast get run on every commit, and only tests that run on every commit actually prevent regressions. Most of the payoff from hexagonal architecture is collected right here.

Pitfalls and Failure Modes

Pitfall 1. The Anemic Domain

The symptom first. The use case class keeps growing. OrderService passes 300 lines and gains another conditional every sprint. Meanwhile the Order entity has not changed in months, and opening it reveals nothing but getters and setters. The class diagram looks like a domain model while every actual rule lives in the service.

Diagnose in this order. First, look for places where the domain entity raises. If there is not a single exception in the entity signalling a rule violation, the rules are not in the entity. Second, count the lines in use cases that read an entity's state field and branch on it. When state-based branching is scattered across several use cases, that judgement belonged to the entity. Third, check whether the same rule is duplicated across two or more use cases. Once you find duplication, the diagnosis is confirmed.

# ❌ Anemic: the judgement lives in the use case
class OrderService:
    def cancel(self, order_id):
        order = self._orders.find_by_id(order_id)
        if order.status == "shipped":          # the rule is here
            raise ValueError("Cannot cancel a shipped order")
        if order.status == "cancelled":        # and again in another use case
            raise ValueError("Already cancelled")
        order.status = "cancelled"             # state mutated from outside
        self._orders.save(order)

# ✅ Move the judgement into the entity
class OrderService:
    def cancel(self, order_id):
        order = self._orders.find_by_id(order_id)
        order.cancel()                          # the entity decides
        self._orders.save(order)

The fix is to find every place that assigns state from outside and move it into an entity method. As long as assignments like order.status = ... survive outside the domain, rules will leak out again. Once the rules gather in the entity, the use case naturally shrinks.

Pitfall 2. The Leaky Port

The symptom shows up the moment you try to swap an adapter. You set out to write an in-memory fake and find that the port's return type is an ORM model, so there is nothing to imitate. Or the port method takes a session object as an argument, so an implementation without a session cannot satisfy the signature in the first place. The instant a port exposes infrastructure types, it stops being a contract and becomes an alias for the ORM.

# ❌ A leaking port — infrastructure surfaces in the signature
class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, session, order_id) -> "OrderTable":   # returns an ORM model
        ...

    @abstractmethod
    def execute_query(self, sql: str) -> list[tuple]:          # storage vocabulary
        ...

# ✅ Keep only domain types and domain vocabulary
class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, order_id: UUID) -> Order | None:
        ...

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

Diagnosis is often a single grep. Search the domain and ports packages for infrastructure imports; if even one line comes back, it is already leaking.

# Do domain/ports import infrastructure? The result must be empty.
grep -rnE '^[[:space:]]*(from|import)[[:space:]]+(sqlalchemy|psycopg2|pymongo|django|fastapi|redis|boto3)' \
     domain/ ports/

# A form you can drop straight into CI (fails on any hit)
if grep -rqE '^[[:space:]]*(from|import)[[:space:]]+(sqlalchemy|psycopg2|django|fastapi)' domain/ ports/; then
  echo "FAIL: domain or ports import infrastructure"
  exit 1
fi

Put this check in CI and the rule lives in the pipeline instead of in someone's memory. Architectural rules that are not checked automatically mostly collapse within a few months.

Pitfall 3. DTO Explosion

Create a new type at every boundary and you end up with four or five classes representing the same order: request DTO, domain entity, persistence model, response DTO, and the payload you send to the payment provider. Adding one field means editing five places, and the mapping code grows longer than the domain code.

This cost is real, and it is not something you must always accept. Two criteria usually settle it. First, do the two types change for different reasons? An API response schema changes because of client needs; a domain model changes because of rules. Different reasons to change means the split earns its keep. Second, is the type crossing that boundary exposed publicly? If a public API response uses the domain entity directly, renaming a domain field becomes a breaking API change.

Conversely, when a layer is internal only and changes for the same reasons the domain does, you may skip the mapping. If the persistence model and the domain entity are effectively the same shape and will clearly keep changing together, maintaining two copies is a principle applied for its own sake. This judgement has to be made again each time; there is no single right answer.

Pitfall 4. Transaction Boundaries

This is the hardest problem in practice. The domain must not know about transactions, yet in the cancellation use case above, saving the order and restocking inventory must commit together. If only one succeeds, stock goes up while the order is still alive, or vice versa.

To be honest about it: there is no universal solution. Three answers are commonly used, and each has a cost.

First, the transaction-script approach: open and close the transaction outside the use case, in the inbound adapter or a decorator. The domain stays completely clean. The cost is that the transaction boundary easily drifts away from the use case boundary. Calling two use cases in one request silently makes them a single transaction, and nothing in the code says so.

Second, making Unit of Work a port: instead of exposing "transaction" as an infrastructure concept, expose "one unit of work" as a domain concept, and let only the adapter know what a commit really is. The cost is that this abstraction leaks easily. The moment you need isolation levels, nested transactions, or savepoints, the port signature starts to resemble a database.

# ports/output/unit_of_work.py — a transaction boundary wrapped in domain vocabulary
from abc import ABC, abstractmethod


class UnitOfWork(ABC):
    """Exposes only the notion of one unit of work.
    What commit and rollback really are is known only to the adapter."""

    @abstractmethod
    def __enter__(self) -> "UnitOfWork":
        ...

    @abstractmethod
    def __exit__(self, exc_type, exc, tb) -> None:
        ...

    @abstractmethod
    def commit(self) -> None:
        ...


# In the use case
    def execute(self, order_id: UUID) -> Order:
        with self._uow:
            order = self._orders.find_by_id(order_id)
            order.cancel()
            self._orders.save(order)
            self._inventory.restock(order.id, order.restock_lines())
            self._uow.commit()

        # The refund cannot join the same transaction (external system)
        # → record it in an outbox and let a separate worker retry
        self._outbox.append("order.cancelled", order.id)
        return order

Third, accepting eventual consistency — that is the last line above. Only what lives in the same database goes into one transaction; calls to external systems are recorded in an outbox table and retried by a separate worker. The cost is explicit: because there are retries, the receiving side must be idempotent, and an intermediate state of "cancelled but not yet refunded" genuinely exists, so operations and support staff have to understand it.

Whichever you choose, nothing is free. That said, external systems such as a payment provider or a WMS cannot participate in your database transaction to begin with, so as soon as two or more external systems are involved, the third answer becomes the only practical option.

When Not to Use This

There are cases where you are clearly better off not using this architecture.

First, a CRUD service with no domain rules. If all it does is accept a request, validate it, put it in one table, and return it, ports and adapters add travel distance rather than safety. Adding one field means editing the request DTO, the domain model, the port signature, the adapter mapping, and the response DTO in sequence. In services like this, a controller talking to the ORM directly is easier to both read and change.

Second, a small team that never swaps adapters. The cost of an indirection layer is paid daily, but the benefit is collected only at the moment of replacement or testing. If that moment never arrives, only the cost remains.

Third, code with a short lifespan. An experimental service or a tool you will use for one quarter has no domain that outlives the frameworks around it. The premise of this architecture simply does not hold.

That gives two preconditions for adoption. One: is there real domain logic that will outlive the frameworks around it? If you can name several rules — "state transition rules", "pricing rules", "approval conditions" — and they change often, the premise holds. Two: does more than one adapter exist per port? Counting a test double as the second adapter is legitimate here; the original write-up itself names running the application in full isolation with mock adapters as the ultimate benefit. But the test has to actually exist before it counts. An interface nobody implements twice is not a second implementation, it is just a file.

Finally, this is not all-or-nothing. Applying ports and adapters only to the modules where the rules cluster — billing, settlement — and leaving the rest as a plain layered design is the most common compromise in practice, and it usually works well.

Conclusion

The core values of Hexagonal Architecture:

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

References

• Alistair Cockburn, "Hexagonal Architecture" (the original Ports and Adapters write-up) — the one-sentence intent, the definitions of primary and secondary actors, the author's position on how many ports to have, and the benefit of running the application in isolation with mock adapters all come from here. https://alistair.cockburn.us/hexagonal-architecture/ (2026-08-16 verified)

The code in this article assumes Python 3.10 or later syntax (union types written as X | None), and the test output example follows the pytest 8.x format. Output formatting differs between versions, so check the actual values in your own environment.

Quiz