Domain-Driven Design & Hexagonal Architecture: Building Decoupled, Testable Python 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

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 (detailed structural safety aspects of dataclasses, enums, and immutability are covered in future posts). ⊕ Using frozen=True dataclasses and returning updated copies via replace() is heavily inspired by Yehonathan Sharvit’s Data-Oriented Programming (DOP) principles of treating data as immutable structures. We will explore this paradigm in depth in Immutability & Safety.

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

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

@dataclass(frozen=True)
class Order:
    id: str
    customer_email: str
    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)

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 key advantages:

• Unidirectional imports: adapters in the infrastructure layer do not need to import anything from the domain or ports layer just to inherit from them. They only need to implement the corresponding method signatures, ensuring that the domain has zero knowledge of the adapter implementations.
• Simplified testing: you can write lightweight test doubles (fakes or stubs) without dragging along abstract base class inheritance boilerplate.
• Separation of concerns: you can easily define narrow protocols that fit specific client needs, adhering to the Interface Segregation Principle.

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

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

@runtime_checkable
class OrderRepositoryPort(RepositoryPort[Order, str], Protocol):
    ...

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

@runtime_checkable
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 (introduced in Python 3.12) 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. Furthermore, 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.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: str) -> bool:
        order = await self.repository.get_by_id(order_id)
        if order is None:
            raise ValueError(f"Order {order_id} not found")

        success = await self.payment_gateway.charge(order)
        if not success:
            return False

        try:
            paid_order = order.mark_as_paid()
            await self.repository.save(paid_order)
            return True
        except Exception:
            await self.payment_gateway.refund(order)
            raise

The Infrastructure Adapter

The Stripe payment adapter implements the payment gateway port and imports external network client libraries like httpx.

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

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:
        try:
            response = await self.client.post(...)
            return response.status_code == 200
        except httpx.HTTPError:
            return False

    async def refund(self, order: Order) -> None:
        try:
            await self.client.post(...)
        except httpx.HTTPError:
            pass

Primitive Obsession in Ports

We can easily fall into the trap of primitive obsession by defining port signatures using raw primitive types rather than rich domain objects. For example, a port defined with primitives bypasses type safety at the compiler level:

# 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, we should pass fully formed domain models directly across our boundaries:

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

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

Leaking Infrastructure Types Upstream

I’ve seen codebases leak infrastructure details upstream, which completely defeats the purpose of architectural boundaries. For example, catching database-specific exceptions inside our application services couples us to a single 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 :

# 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 and keeps our error handling 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. By defining ports using typing.Protocol, we rely instead on structural typing 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, allowing us to build lightweight stubs and decoupled adapters with ease.

Fragile Mocking vs In-Memory Fakes

Relying heavily on mocking libraries 1010 like unittest.mock.AsyncMock inside a test suite makes our tests fragile. Mocks simulate behaviour but won’t warn us 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
from shop.ports.repository import OrderRepositoryPort

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

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

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

Injecting this fake adapter into our service tests lets us verify the orchestration flow of our application services without network overhead, database configuration, or fragile mock setups.