Ports in Hexagonal Architecture on Python — how the core talks to the outside world

← Back to the section

The application core holds the business logic. But almost any business logic sooner or later reaches into the outside world: save an aggregate to the database, start a payment, publish an event. The problem is that if the core directly calls SQLAlchemy or an external HTTP client, then the core starts depending on a concrete piece of infrastructure. Change the payment system — and you have to dig into the business logic. Writing a test without spinning up a DB becomes impossible.

A port is the solution. The core declares an interface: "I need something that can save an order" or "I need something that can register a payment". Who exactly does it — the core doesn't care. The implementation (the adapter) is supplied by the DI container at startup.

Where a port lives

A port is a Protocol in core/<bounded-context>/port/out/. Exactly in core/, not in adapters/. The contract belongs to the core, not to whoever implements it.

src/<service>/
  core/
    orders/
      aggregate/order.py
      port/
        out/
          order_repository.py       # persistence (aggregate)
          order_view_repository.py  # read projection (CQRS read-side)
          payment_port.py           # external HTTP system
          notification_port.py      # SMS / email
          order_event_publisher.py  # outbound events
  adapters/
    out/
      postgres/                     # implementation of order_repository.py
      sber/                         # implementation of payment_port.py

The naming convention:

Repository without the Port suffix is a convention from DDD — the name speaks for itself. Everything else gets the Port suffix, to make it visible: this is a contract to an external system.

How to declare a port through a Protocol

A port is always a Protocol (or an ABC), not an ordinary class. An ordinary class would require inheritance, which would couple the adapter to the core. A Protocol works structurally: the adapter isn't required to import the Protocol — matching signatures is enough.

# core/orders/port/out/order_repository.py
from typing import Protocol
from core.orders.aggregate.order import Order
from core.orders.value_object.order_id import OrderId


class OrderRepository(Protocol):
    async def find_by_id(self, order_id: OrderId) -> Order | None: ...
    async def find_required(self, order_id: OrderId) -> Order: ...  # raises OrderNotFoundError
    async def save(self, order: Order) -> None: ...

# core/orders/port/out/payment_port.py
from typing import Protocol
from core.orders.command.register_payment_command import RegisterPaymentCommand
from core.orders.value_object.payment_id import PaymentId
from core.orders.dto.register_result import RegisterResult


class PaymentPort(Protocol):
    async def register(self, cmd: RegisterPaymentCommand) -> RegisterResult: ...
    async def cancel(self, payment_id: PaymentId) -> None: ...

Adding @runtime_checkable isn't necessary — static analysis is enough. Add it only if somewhere at runtime you do isinstance(adapter, PaymentPort).

A port's methods work with domain types

This is the most common mistake when moving to Hexagonal. A port must accept and return types from the domain, not types from a concrete external system.

# Correct — domain types in the signature
class PaymentPort(Protocol):
    def register(self, cmd: RegisterPaymentCommand) -> RegisterResult: ...
    #                        ↑ domain command           ↑ domain result

# Mistake — Sber DTO in the port signature
class PaymentPort(Protocol):
    def register(self, req: SberRegisterRequest) -> SberRegisterResponse: ...
    #                        ↑ Sber-SDK detail       ↑ Sber-SDK detail

Why the second variant breaks the architecture:

• The core starts depending on Sber's details. Moving to another payment provider requires changes in the business logic.
• Tests on the handlers are forced to create a SberRegisterRequest — an infrastructure object in a unit test.
• The Sber API schema (fields, formats) seeps into the core, even though this is knowledge about the infrastructure.

PaymentPort accepts a RegisterPaymentCommand with domain fields (amount, order_id, description) and returns a RegisterResult with domain fields (payment_id, redirect_url). Translation into Sber's formats lies in SberPaymentAdapter in the adapters/out/sber/ package — the core doesn't know about it.

Exceptions: domain ones in core, concrete ones in adapters

A port's exceptions are declared in core/ next to the port itself. The adapter may raise their subclasses — the handler in the core catches the base one, without knowing about the concrete adapter.

# core/orders/port/out/payment_port.py
class PaymentPortError(Exception):
    """Base exception for any payment adapter."""


class PaymentNotFoundError(PaymentPortError):
    def __init__(self, payment_id: PaymentId) -> None:
        super().__init__(f"Payment not found: {payment_id}")
        self.payment_id = payment_id


class PaymentDeclinedError(PaymentPortError):
    def __init__(self, payment_id: PaymentId, reason: str) -> None:
        super().__init__(f"Payment {payment_id} declined: {reason}")
        self.payment_id = payment_id
        self.reason = reason

# adapters/out/sber/sber_error.py
from core.orders.port.out.payment_port import PaymentPortError


class SberError(PaymentPortError):
    """A concrete Sber adapter error — an adapter detail, not the core's."""

# adapters/out/sber/sber_payment_adapter.py
import httpx
from core.orders.port.out.payment_port import PaymentPort, PaymentPortError
from adapters.out.sber.sber_error import SberError


class SberPaymentAdapter:
    def __init__(self, client: httpx.AsyncClient) -> None:
        self._client = client

    async def register(self, cmd: RegisterPaymentCommand) -> RegisterResult:
        try:
            resp = await self._client.post("/register", json=self._to_sber_request(cmd))
            resp.raise_for_status()
            return self._to_domain_result(resp.json())
        except httpx.HTTPError as exc:
            raise SberError("Sber register failed") from exc

The handler in the core catches the domain exception, not the specific one:

# core/orders/usecase/create_payment_handler.py
class CreatePaymentHandler:
    def __init__(self, payment_port: PaymentPort) -> None:
        self._payment_port = payment_port

    async def handle(self, cmd: CreatePaymentCommand) -> Payment:
        try:
            result = await self._payment_port.register(RegisterPaymentCommand(...))
            return Payment(payment_id=result.payment_id, ...)
        except PaymentDeclinedError as exc:
            raise OrderPaymentDeclinedError(cmd.order_id, exc.reason) from exc
        except PaymentPortError as exc:
            raise PaymentSystemUnavailableError() from exc

Replace SberPaymentAdapter with OdnaKassaPaymentAdapter — the handler doesn't need to be rewritten.

Inbound port: how requests reach the core

In classic Hexagonal there's also an inbound port (the entry) — an interface that the core exposes outward. In the UCP approach on Python, the role of the inbound port is played by a UseCase + Handler paired with a Dispatcher.

The FastAPI router doesn't call CreateOrderHandler directly. It passes the command to the Dispatcher, which itself finds the right handler by the command type:

# adapters/in_/http/orders/order_router.py
from fastapi import APIRouter, Depends
from core.shared.dispatcher import Dispatcher
from core.orders.command.create_order_command import CreateOrderCommand
from adapters.in_.http.orders.order_request_mapper import to_command, to_response

router = APIRouter()


@router.post("/orders", status_code=201)
async def create_order(
    body: CreateOrderRequest,
    dispatcher: Dispatcher = Depends(),
) -> OrderResponse:
    cmd = to_command(body)
    order = await dispatcher.dispatch(cmd)
    return to_response(order)

The handler accepts port interfaces, not concrete adapters:

# core/orders/usecase/create_order_handler.py
class CreateOrderHandler:
    def __init__(
        self,
        order_repository: OrderRepository,
        product_repository: ProductRepository,
        customer_repository: CustomerRepository,
    ) -> None:
        self._orders = order_repository
        self._products = product_repository
        self._customers = customer_repository

    async def handle(self, cmd: CreateOrderCommand) -> Order:
        customer = await self._customers.find_required(cmd.customer_id)
        products = [await self._products.find_required(pid) for pid in cmd.product_ids]
        order = Order.create(customer=customer, products=products)
        await self._orders.save(order)
        return order

Common mistakes

The port sits in the adapter's folder. The port is a contract of the core, it should be in core/, not in adapters/out/sber/.

An external system's DTO in the port signature. Instead of SberRegisterRequest — the domain command RegisterPaymentCommand. Mapping into the system's formats is the adapter's job.

find_by_id returns Order | None where absence is an error. Better an explicit method find_required that raises OrderNotFoundError. Handling None in every handler is a repetitive pattern.

The port declared as an ordinary class. An ordinary class requires inheritance and makes substitution in tests cumbersome. A Protocol or ABC — the adapter implements the contract without importing from the core.

No import-linter in CI. Python doesn't check package boundaries at the compiler level. Without import-linter (a layers contract in pyproject.toml) someone will sooner or later import SQLAlchemy right into core/ — and a test won't catch it.

In short

• A port is a Protocol in core/<bc>/port/out/. The core describes what it needs; the adapter implements it.
• Naming: <X>Repository for persistence, <Y>Port for external HTTP systems, <Z>EventPublisher for events.
• A port's methods accept and return domain types, not an external system's DTOs.
• A port's exceptions live in core/. The adapter raises a subclass; the handler catches the base one.
• The inbound port is implemented by a UseCase + Handler pair, with the Dispatcher as the entry point instead of an explicit InboundPort interface.
• A Protocol instead of an ordinary class — then the adapter isn't required to inherit from anything in the core.
• import-linter in CI is the only protection against boundary violations in Python.

What to read next

• Adapters out — who implements the Protocol port from the core.
• Adapters in — how a FastAPI router uses the Dispatcher.
• Composition root — how adapters are supplied to ports at startup.
• Core layer — what is allowed and forbidden in the core.
• Architecture tests — the import-linter contract.
• Module structure — the core / adapters / app package layout.