Ports в Hexagonal Architecture на Python — как ядро говорит с внешним миром

← назад к разделу

Ядро приложения хранит бизнес-логику. Но почти любая бизнес-логика рано или поздно обращается во внешний мир: сохранить агрегат в базу, запустить платёж, опубликовать событие. Проблема в том, что если ядро напрямую вызывает SQLAlchemy или внешний HTTP-клиент, то ядро начинает зависеть от конкретной инфраструктуры. Поменять платёжную систему — нужно лезть в бизнес-логику. Написать тест без поднятия БД — нельзя.

Port — это решение. Ядро объявляет интерфейс: «мне нужно что-то, что умеет сохранять заказ» или «мне нужно что-то, что умеет регистрировать платёж». А кто именно это делает — ядру не важно. Реализацию (адаптер) подкладывает DI-контейнер на старте.

Где живёт port

Port — это Protocol в core/<bounded-context>/port/out/. Именно в core/, не в adapters/. Контракт принадлежит ядру, а не тому, кто его реализует.

src/<service>/
  core/
    orders/
      aggregate/order.py
      port/
        out/
          order_repository.py       # persistence (агрегат)
          order_view_repository.py  # read-проекция (CQRS read-side)
          payment_port.py           # внешняя HTTP-система
          notification_port.py      # SMS / email
          order_event_publisher.py  # исходящие события
  adapters/
    out/
      postgres/                     # реализация order_repository.py
      sber/                         # реализация payment_port.py

Соглашение по именам:

Тип port'а	Имя	Пример
Persistence write	`<X>Repository`	`OrderRepository`
Persistence read-проекция	`<X>ViewRepository`	`OrderViewRepository`
Внешняя HTTP-система	`<Y>Port`	`PaymentPort`, `SmsPort`
Исходящие события	`<Z>EventPublisher`	`OrderEventPublisher`

Repository без суффикса Port — соглашение из DDD, название само говорит за себя. Всё остальное получает суффикс Port, чтобы было видно: это контракт к внешней системе.

Как объявить port через Protocol

Port всегда Protocol (или ABC), а не обычный класс. Обычный класс потребует наследования, что свяжет адаптер с ядром. Protocol работает структурно: адаптер не обязан импортировать Protocol — достаточно совпадения сигнатур.

# 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: ...  # бросает 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: ...

@runtime_checkable добавлять не нужно — статического анализа достаточно. Добавьте его только если где-то в рантайме делаете isinstance(adapter, PaymentPort).

Методы port'а работают с domain-типами

Это самая частая ошибка при переходе на Hexagonal. Port должен принимать и возвращать типы из домена, а не типы от конкретной внешней системы.

# Правильно — domain-типы в сигнатуре
class PaymentPort(Protocol):
    def register(self, cmd: RegisterPaymentCommand) -> RegisterResult: ...
    #                        ↑ domain command           ↑ domain result

# Ошибка — DTO Сбера в сигнатуре порта
class PaymentPort(Protocol):
    def register(self, req: SberRegisterRequest) -> SberRegisterResponse: ...
    #                        ↑ деталь Sber-SDK      ↑ деталь Sber-SDK

Почему второй вариант ломает архитектуру:

Ядро начинает зависеть от деталей Сбера. Переход на другую платёжку требует правок в бизнес-логике.
Тесты на handler'ах вынуждены создавать SberRegisterRequest — инфраструктурный объект в unit-тесте.
Схема Sber-API (поля, форматы) просачивается в core, хотя это знание об инфраструктуре.

PaymentPort принимает RegisterPaymentCommand с доменными полями (amount, order_id, description) и возвращает RegisterResult с доменными полями (payment_id, redirect_url). Перевод в форматы Сбера лежит в SberPaymentAdapter в пакете adapters/out/sber/ — ядро об этом не знает.

Исключения: доменные в core, конкретные в адаптерах

Исключения port'а объявляют в core/ рядом с самим port'ом. Адаптер может бросать их подклассы — handler в ядре ловит базовый, не зная о конкретном адаптере.

# core/orders/port/out/payment_port.py
class PaymentPortError(Exception):
    """Базовое исключение для любого payment-адаптера."""


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):
    """Конкретная ошибка Sber-адаптера — деталь адаптера, не ядра."""

# 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

Handler в ядре ловит доменное исключение, не специфическое:

# 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

Замените SberPaymentAdapter на OdnaKassaPaymentAdapter — handler переписывать не нужно.

Inbound port: как запросы попадают в ядро

В классическом Hexagonal есть и inbound port (вход) — интерфейс, который ядро предоставляет наружу. В UCP-подходе на Python роль inbound port играет UseCase + Handler в паре с Dispatcher.

FastAPI-роутер не вызывает CreateOrderHandler напрямую. Он передаёт команду в Dispatcher, а тот сам находит нужный handler по типу команды:

# 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)

Handler принимает port-интерфейсы, не конкретные адаптеры:

# 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

Частые ошибки

Port лежит в папке адаптера. Порт — контракт ядра, он должен быть в core/, а не в adapters/out/sber/.

В сигнатуре port'а — DTO внешней системы. Вместо SberRegisterRequest — доменная команда RegisterPaymentCommand. Маппинг в форматы системы — задача адаптера.

find_by_id возвращает Order | None там, где отсутствие — это ошибка. Лучше явный метод find_required, который бросает OrderNotFoundError. Обработка None в каждом handler'е — повторяющийся шаблон.

Port объявлен как обычный класс. Обычный класс требует наследования и делает подмену в тестах громоздкой. Protocol или ABC — адаптер реализует контракт без импорта из ядра.

Нет import-linter в CI. Python не проверяет границы пакетов на уровне компилятора. Без import-linter (контракт layers в pyproject.toml) кто-нибудь рано или поздно импортирует SQLAlchemy прямо в core/ — и тест это не поймает.

Коротко

Port — Protocol в core/<bc>/port/out/. Ядро описывает, что ему нужно; реализует адаптер.
Именование: <X>Repository для persistence, <Y>Port для внешних HTTP-систем, <Z>EventPublisher для событий.
Методы port'а принимают и возвращают domain-типы, не DTO внешней системы.
Исключения port'а живут в core/. Адаптер бросает подкласс; handler ловит базовый.
Inbound port реализует пара UseCase + Handler, Dispatcher — точка входа вместо явного InboundPort-интерфейса.
Protocol вместо обычного класса — тогда адаптер не обязан наследоваться от чего-то из ядра.
import-linter в CI — единственная защита от нарушения границ в Python.

Что почитать дальше

Adapters out — кто реализует Protocol-порт из core.
Adapters in — как FastAPI-роутер использует Dispatcher.
Composition root — как подкладывают адаптеры на порты при старте.
Core layer — что разрешено и запрещено в core.
Архитектурные тесты — import-linter контракт.
Структура модулей — пакетная раскладка core / adapters / app.