Hexagonal Architecture in Python: Building Decoupled, Testable Systems

Directory mapping

A clean Hexagonal directory structure enforces these import boundaries:

shop/
├── domain/
│   ├── __init__.py
│   └── order.py         # Pure domain entities, zero external imports
├── ports/
│   ├── __init__.py
│   ├── repository.py    # Outbound Protocols (e.g., OrderRepositoryPort)
│   └── gateway.py       # Outbound Protocols (e.g., PaymentGatewayPort)
├── application/
│   ├── __init__.py
│   └── service.py       # Inbound Driving Port / Service Orchestrator
└── adapters/
    ├── __init__.py
    ├── database.py      # Postgres/SQLAlchemy adapter implementation
    └── payment.py       # Stripe HTTP client adapter implementation

Alistair Cockburn’s 2005 article uses the terms primary and secondary ports/adapters, not driving and driven; the latter was popularized by Vaughn Vernon in Implementing Domain-Driven Design (2013) and is the terminology I like to use. Cockburn also explicitly says the hexagon shape is a visual choice, not a count constraint: his typical recommendation is 2-4 ports per service.1313 Cockburn, Alistair. Hexagonal Architecture (2005), “Ports and Adapters” More than that, and you have a port proliferation problem (the 5th anti-pattern below) that is the most common reason hexagonal codebases collapse.

One file per concrete adapter is the right convention. Two implementations of the same port (e.g., StripePaymentAdapter and BraintreePaymentAdapter) belong in separate files even if they sit in the same adapters/payment/ directory.

The pure domain layer

The order domain module contains only pure Python code and standard typing features.

To keep the domain layer clean and decoupled, the design relies strictly on standard library constructs rather than third-party frameworks. In this case, the domain models use a combination of enums and dataclasses.

from dataclasses import dataclass, replace
from decimal import Decimal
from enum import StrEnum, auto
from typing import NewType

OrderId = NewType("OrderId", str)
CustomerEmail = NewType("CustomerEmail", str)

class OrderStatus(StrEnum):
    PENDING = auto()
    PAID = auto()
    FAILED = auto()

@dataclass(frozen=True, slots=True)
class Order:
    id: OrderId
    customer_email: CustomerEmail
    total_amount: Decimal
    status: OrderStatus = OrderStatus.PENDING

    def mark_as_paid(self) -> "Order":
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot pay order in status {self.status}")
        return replace(self, status=OrderStatus.PAID)

NewType is a static-only construct. At runtime OrderId("o-1") returns "o-1" unchanged (docs.python.org/3/library/typing.html#newtype). If the value comes from an untrusted source (an HTTP body, a Kafka message, a Redis payload), validate the value at the adapter boundary using pydantic.dataclasses.dataclass(frozen=True) or an Annotated[Type, StringConstraints(...)] rather than relying on the NewType for runtime safety. The NewType is a static type-check aid; the boundary validation is what makes the value trustable.

The driven ports

The driven port protocols define the contracts that infrastructure adapters must implement, using structural subtyping via typing.Protocol 66 ports/repository.py and ports/gateway.py .

In Python, Protocols implement structural subtyping (often called static duck typing). In contrast to traditional nominal inheritance (where a class must explicitly inherit from a parent class like abc.ABC), structural typing only requires that a class implements the attributes and methods specified by the Protocol. If it matches the shape, the type checker is satisfied.

Using Protocols for driven ports offers three advantages over nominal inheritance. Adapters in the infrastructure layer do not need to import anything from the domain or ports layer just to inherit (they only need to implement the corresponding method signatures, so the domain has zero knowledge of the adapter implementations). Test code can use lightweight fakes or stubs instead of dragging along ABC boilerplate, and narrow protocols that fit specific client needs are easier to express than under nominal subtyping.

from typing import Protocol
from shop.domain.order import Order, OrderId

class RepositoryPort[T, ID](Protocol):
    async def save(self, entity: T) -> None: ...
    async def get_by_id(self, entity_id: ID) -> T | None: ...

class OrderRepositoryPort(RepositoryPort[Order, OrderId], Protocol):
    ...

from typing import Protocol
from shop.domain.order import Order

class PaymentGatewayPort(Protocol):
    async def charge(self, order: Order) -> bool: ...
    async def refund(self, order: Order) -> None: ...

Under the hood, the generic RepositoryPort[T, ID] uses PEP 695 1111 Python 3.12 Generics to declare type parameters:

• Generic type parameters [T, ID]: this syntax declares T and ID as generic type parameters scoped directly to the class statement, eliminating the legacy TypeVar and Generic[...] boilerplate.
• Type bounds: you can restrict type parameters by specifying an upper bound, e.g., class RepositoryPort[T: DomainEntity, ID: UUID | str](Protocol). This is conceptually similar to Rust’s generic bounds (e.g., T: Entity), ensuring that only subtypes of DomainEntity or specific ID types can satisfy the generic contract.
• Scope and advantages: aside from dramatically cleaner syntax, the compiler scopes type variables strictly to the class block rather than polluting the module namespace, and type checkers can automatically infer variance. ⊕ Variance describes how subtyping of a component type (e.g., Dog is an Animal) affects the subtyping of the container type (e.g., Container[Dog] vs Container[Animal]). It can be covariant (preserves subtyping), contravariant (reverses it), or invariant (requires an exact type match).
• Constraints and limitations: Python bounds cannot express multiple independent protocols (e.g., no direct equivalent to Rust’s T: Display + Clone intersection bounds) unless you declare a single compound protocol that inherits from both. Static type checkers (like Pyright or MyPy) check these bounds; standard Python does not enforce them at runtime.
• Untyped fallback: if the design omits the type parameters entirely, the port falls back to operating on Any, which disables static check safety and defeats the purpose of structural contracts.

The application service

The application service acts as the driving port orchestrator. The service coordinates database checks and payment triggers strictly using the abstractions.

from shop.domain.order import OrderId
from shop.ports.repository import OrderRepositoryPort
from shop.ports.gateway import PaymentGatewayPort

class OrderProcessorService:
    def __init__(
        self,
        repository: OrderRepositoryPort,
        payment_gateway: PaymentGatewayPort,
    ) -> None:
        self.repository = repository
        self.payment_gateway = payment_gateway

    async def process_payment(self, order_id: OrderId) -> bool:
        order = await self.repository.get_by_id(order_id)
        if order is None:
            raise ValueError(f"Order {order_id} not found")

        await self.payment_gateway.charge(order)
        await self.repository.save(order.mark_as_paid())

The process_payment logic shows a naive implementation with the purpose of showing the hexagonal architecture concepts.

The infrastructure adapter

The Stripe payment adapter implements the payment gateway port and imports external network client libraries like httpx. The AsyncClient is injected (not constructed inside the adapter) so that the composition root1212 the single place in the application where adapters are constructed and wired into ports, typically main.py or a bootstrap_application function can configure timeouts, connection-pool limits, and base URL once and share the pool across adapters.

import httpx
import msgspec
from shop.domain.order import Order
from shop.ports.gateway import PaymentGatewayPort

class ChargeRequest(msgspec.Struct):
    amount_cents: int
    currency: str = "usd"

class StripePaymentAdapter(PaymentGatewayPort):
    def __init__(self, api_key: str, client: httpx.AsyncClient) -> None:
        self.api_key = api_key
        self.client = client

    async def charge(self, order: Order) -> bool:
        body = msgspec.to_builtins(ChargeRequest(amount_cents=int(order.total_amount * 100)))
        try:
            response = await self.client.post(
                "https://api.stripe.com/v1/charges",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json=body,
            )
            return response.status_code == 200
        except httpx.HTTPError:
            return False

This adapter shows two patterns:

• Thin vs thick adapters. The StripePaymentAdapter above is a thin adapter: it translates transport, not business logic. A thick adapter is one where the external party owns the contract and the business logic lives in the adapter (e.g., a tax-calculation API, an OAuth flow, a payment gateway with a complex state machine). For thick adapters, the port should expose intent (TaxGatewayPort.calculate(order) -> TaxBreakdown) rather than transport (HTTPClientPort.post(...)). Chris Richardson distinguishes domain services (pure business logic) from application services (use case orchestration) in Microservices Patterns; adapters that contain domain logic are the third category.
• Boundary validation with msgspec or Pydantic v2. Parsing the response into a msgspec.Struct (or pydantic.BaseModel) before returning a domain entity gives you validation that a dict or httpx.Response.json() cannot. Without it, a renamed Stripe field (amount → amount_cents) silently produces a None deep in the domain layer.

Primitive obsession in ports

Primitive obsession occurs when port signatures are defined using raw primitive types rather than rich domain objects. For example, a port defined with primitives bypasses type safety at the compiler level. The type checker sees two str arguments and stays silent:

# Anti-pattern: primitive obsession in port signatures
async def save(self, order_id: str, total_amount: float) -> None: ...

This configuration forces adapters to handle database validation logic that belongs in the domain. Instead, ports should accept fully formed domain models directly across the boundaries, so the invariants are checked once, in the domain, before any persistence attempt:

# Domain-centric port signature
async def save(self, order: Order) -> None: ...

Passing the Order model ensures that business invariants are validated inside the domain boundaries before any persistence attempt.

Leaking infrastructure types upstream

Leaking infrastructure details upstream defeats the purpose of architectural boundaries. Catching database-specific exceptions inside application services, for example, couples the service to a specific persistence technology:

# Anti-pattern: catching infrastructure-specific exceptions in service layers
try:
    await self.repository.save(order)
except SQLAlchemyError as e:  # Couples service directly to SQLAlchemy
    raise ApplicationError("Database update failed") from e

To maintain boundary isolation, database adapters must catch their own technology-specific exceptions internally and raise custom domain exceptions 77 e.g., DomainRepositoryError or return None (if it makes sense) instead:

# Proper pattern: adapters map exceptions to custom domain exceptions
try:
    await self.session.commit()
except SQLAlchemyError as e:
    raise DomainRepositoryError("Database persistence failed") from e

Raising custom domain exceptions shields the application service from database details. Error handling stays independent of the infrastructure.

Overusing nominal class inheritance

If you come from OOP languages, you might be used to overusing nominal class inheritance by defining ports as abstract base classes (abc.ABC). Nominal inheritance 88 an adapter must explicitly inherit from the ABC forces adapters to import and subclass the interface explicitly:

# Nominal inheritance coupling
from shop.ports.repository import OrderRepositoryABC

class PostgresRepository(OrderRepositoryABC): ...

This creates rigid inheritance chains and complicates testing. Defining ports with typing.Protocol relies on structural typing instead 99 any class with matching method signatures satisfies it :

# Structural subtyping (no imports needed in the adapter file)
class PostgresRepository:
    # Naturally conforms to OrderRepositoryPort by implementing the required methods
    async def save(self, order: Order) -> None: ...

Any class with matching signatures automatically satisfies the port without explicit inheritance, making it straightforward to build lightweight stubs and decoupled adapters.

Port proliferation

The most common reason hexagonal codebases collapse is too many ports, not too few. When every external dependency becomes a Protocol and every Protocol has 1-2 methods, the application service ends up with 8-12 constructor parameters, the test file has 8-12 in-memory fakes, and any refactor touches every layer. The cure is the Interface Segregation Principle (Robert C. Martin, Clean Architecture): combine ports that travel together into one Role Interface.

The test: “am I likely to need a second implementation of this method?” If the answer is no, the method does not need its own port.

# Anti-pattern: one method per port
class UserRepositoryPort(Protocol):
    async def save(self, user: User) -> None: ...
class UserCachePort(Protocol):
    async def put(self, user: User) -> None: ...
class UserSearchPort(Protocol):
    async def find(self, query: str) -> list[User]: ...

# Better: one Role Interface that travels together
class UserReadModelPort(Protocol):
    async def save(self, user: User) -> None: ...
    async def put(self, user: User) -> None: ...
    async def find(self, query: str) -> list[User]: ...

Three single-method ports become one three-method port, one adapter, and one fake in the test suite. Of course, this is an obvious example. In real life you will have to think very carefully where to draw these boundaries.

Fragile mocking vs in-memory fakes

Relying heavily on mocking libraries 1010 like unittest.mock.AsyncMock inside a test suite makes tests fragile. Mocks simulate behavior but do not warn the developer if a port’s method signature changes, leading to false green results in CI when the real code is broken.

An alternative approach is writing an in-memory fake adapter for testing. A fake is a lightweight, zero-dependency class that simulates a database or gateway using simple local Python dictionaries or lists:

from shop.domain.order import Order, OrderId
from shop.ports.repository import OrderRepositoryPort

class InMemoryOrderRepository(OrderRepositoryPort):
    def __init__(self) -> None:
        self._orders: dict[OrderId, Order] = {}

    async def save(self, order: Order) -> None:
        self._orders[order.id] = order

    async def get_by_id(self, order_id: OrderId) -> Order | None:
        return self._orders.get(order_id)

Injecting this fake adapter into service tests verifies the orchestration flow of the application service without network overhead, database configuration, or fragile mock setups.