Ports в Hexagonal Architecture — Protocol в core/ (Python)

Опирается на правила: R-HEX-PORT-1 … R-HEX-PORT-4 и R-HEX-PORT-X1 … R-HEX-PORT-X4 из Hexagonal Style Guide → раздел 4. Ports.

Важно знать

• Outbound port — Protocol в core/<bc>/port/out/. Описывает, что core нужно от внешнего мира.
• Имена: <X>Repository (persistence), <X>ViewRepository (CQRS read-side), <Y>Port (внешние HTTP-системы), <Z>EventPublisher (события).
• Методы port'а оперируют domain-типами, не DTO внешней системы (SberRegisterRequest — деталь адаптера).
• Port-исключения объявлены в core/ (PaymentPortError); подклассы (SberError) — в out-adapter; handler ловит базовый.
• Inbound port = UseCase + Handler. Отдельного «InboundPort» нет — вход в core идёт через Dispatcher.
• Port — всегда Protocol или ABC, не класс. Класс убивает подмену в тестах.
• import-linter — единственный enforcement: без него нарушение границ останется незамеченным.

В Hexagonal стрелка зависимостей: app → adapters → core. Чтобы это работало, core не должен знать про адаптеры. Но core нужно ходить в БД, дёргать платёжку, публиковать события. Решение — port-интерфейсы в core/. Core описывает что ему нужно, adapter в своём пакете реализует контракт, DI на старте подкладывает реализацию.

В Python нет compile-time изоляции модулей, поэтому Protocol + import-linter — не украшение, а единственный механизм enforcement. Без контракта в pyproject.toml кто-нибудь неизбежно импортнёт SQLAlchemy в core/ и тест это не поймает.

Где живёт port и как он называется

R-HEX-PORT-1: outbound port — Protocol в core/<bc>/port/out/.

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

Конвенция имён:

<X>Repository без суффикса Port — соглашение из DDD; «repository» самоговорящее. Для всего остального — суффикс Port, чтобы было видно: «это контракт к внешней системе».

# 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):
    def find_by_id(self, order_id: OrderId) -> Order | None: ...
    def find_required(self, order_id: OrderId) -> Order: ...  # throws OrderNotFoundError
    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):
    def register(self, cmd: RegisterPaymentCommand) -> RegisterResult: ...
    def cancel(self, payment_id: PaymentId) -> None: ...

Protocol без @runtime_checkable — достаточно для статического анализа и DI через isinstance-проверки не нужны. @runtime_checkable добавляется только если нужно isinstance(adapter, PaymentPort) в runtime.

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

R-HEX-PORT-2: ни DTO внешней системы, ни ORM-модели в сигнатуре не появляются.

# ХОРОШО — 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

Что не так с вариантом «плохо»:

• Core знает про Сбер. Domain-типов больше нет; вместо них — структуры из Sber-SDK. Переход на другую платёжку (OdnaKassa) требует правки везде, где SberRegisterRequest упомянут.
• Тестировать handler нужно через SDK. Тесты на handler'ах создают SberRegisterRequest для каждого сценария — это инфраструктурная деталь в unit-тесте.
• Маппинг просочился в core. SberRegisterRequest строится по схеме Sber-API с его полями и форматами. Это знание о внешней системе, а не доменное.

PaymentPort принимает доменный RegisterPaymentCommand (amount, order_id, description) и возвращает доменный RegisterResult (payment_id, redirect_url). Маппинг в Sber-API лежит в SberPaymentAdapter в пакете adapters/out/sber/.

Port-исключения — иерархия

R-HEX-PORT-3: абстрактные исключения объявляются в core/, конкретные — в out-adapter'ах.

# 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-адаптера — деталь out-adapter, не core."""

# 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

    def register(self, cmd: RegisterPaymentCommand) -> RegisterResult:
        try:
            resp = 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 ловит доменное исключение, не специфическое:

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

    def handle(self, cmd: CreatePaymentCommand) -> Payment:
        try:
            result = 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 не правится. Меняется только класс конкретного исключения в out-adapter.

Inbound port = UseCase

R-HEX-PORT-4: отдельного «InboundPort» нет. В UCP UseCase + Handler — это вход в core, а Dispatcher играет роль того, что в классическом Hexagonal называют InboundPort.

# 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)      # ← inbound-вход в core
    return to_response(order)

Роутер не зовёт CreateOrderHandler напрямую. Он зовёт dispatcher, который сам находит нужный handler по типу команды. Это убирает прямую зависимость роутера от каждого конкретного handler'а.

# 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 = self._customers.find_required(cmd.customer_id)
        products = [self._products.find_required(pid) for pid in cmd.product_ids]
        order = Order.create(customer=customer, products=products)
        self._orders.save(order)
        return order

Handler инжектит port-интерфейсы (OrderRepository, ProductRepository, CustomerRepository), не конкретные адаптеры. DI в app/container.py подкладывает реализации на старте.

Что запрещено

Куда дальше

• Adapters out — кто реализует Protocol-порт из core/.
• Adapters in — как FastAPI-роутер использует Dispatcher как inbound-вход.
• Composition root — как app/container.py подкладывает адаптеры на порты.
• Core layer — что разрешено и запрещено в core/.
• Архитектурные тесты — import-linter контракт и почему он обязателен.
• Структура модулей — пакетная раскладка core / adapters / app.
• Когда применять Hexagonal — признаки, что пора переходить на Уровень 3.