Adapters Out в Hexagonal (Python) — реализация port-Protocol из core

Out-адаптеры в Python Hexagonal: пакет per-system, класс реализует port-Protocol из core, маппер domain ↔ DTO внешней системы, запреты R-HEX-AOUT-X1..X4.

Опубликовано 20 июня 2026 ~6 мин чтения Автор: Вадим Викулин

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

Важно знать

На каждую внешнюю систему — отдельный пакет adapters/out/<system>/: persistence/ (SQLAlchemy), sber/ (httpx), kafka/, s3/.

Класс адаптера реализует port-Protocol из core/. Protocol — контракт, адаптер — реализация.

Маппер в пакете адаптера переводит между domain-типами (сигнатура порта) и DTO внешней системы.

Out-адаптер знает свою инфраструктуру (SQLAlchemy / httpx / aiokafka), не знает другие адаптеры.

Один адаптер реализует один порт конкретного домена. Per-system isolation (R-RES-ISO-1).

Возврат DTO внешней системы из порт-метода — запрещено. Только domain-результат.

Бизнес-логика в out-адаптере — запрещена. Адаптер мапит и вызывает, не решает.

Out-адаптер, инжектирующий другой out-адаптер — антипаттерн. Координация — use case в core/.

Out-адаптер — «выход» из сервиса во внешний мир. Он принимает domain-вызов (payment_port.register(cmd)), маппит в формат внешней системы (Sber API request), вызывает её через httpx/SQLAlchemy/aiokafka, получает ответ, маппит обратно в domain (RegisterResult). На этом ответственность заканчивается — никакой бизнес-логики, никакого «если статус Sber'а такой, то делаем то». Раскрытие правил R-HEX-AOUT-* ниже.

Per-system пакеты

R-HEX-AOUT-1: на каждую внешнюю систему — отдельный пакет adapters/out/<system>/.

src/order_service/
  adapters/out/
    persistence/        # implements OrderRepository через SQLAlchemy (PG)
    sber/               # implements PaymentPort через Sber REST API (httpx)
    ok/                 # implements PaymentPort через OdnaKassa (httpx)
    sms/                # implements SmsPort (httpx)
    kafka/              # implements EventPublisher (aiokafka)
    s3/                 # implements StoragePort (aiobotocore)

Почему именно так:

Изоляция зависимостей. adapters/out/sber/ зависит от Sber-специфичного кода; adapters/out/sms/ — от SMSC-клиента. Замена SMS-провайдера — правка в одном пакете, остальные не трогаются.
Per-system resilience. Tenacity Retry, Circuit Breaker (pybreaker), тайм-ауты httpx — настраиваются отдельно для каждой системы. Общий httpx-клиент на все исходящие вызовы — единая точка отказа.
Per-system observability. Метрики payment_sber_* и sms_smsc_* — разные. Не смешивать в один счётчик.
Per-system testability. respx для Sber поднимается в тестах adapters/out/sber/; respx для SMSC — в тестах adapters/out/sms/. Не один большой mock для всего.

import-linter фиксирует, что адаптеры не зависят друг от друга:

# pyproject.toml
[[tool.importlinter.contracts]]
name = "adapters-independence"
type = "independence"
modules = [
    "order_service.adapters.out.sber",
    "order_service.adapters.out.ok",
    "order_service.adapters.out.persistence",
    "order_service.adapters.out.sms",
]

Адаптер реализует port-Protocol

R-HEX-AOUT-2: класс адаптера реализует port-Protocol из core/.

Порт объявлен в core/:

# core/order/port/out/payment_port.py
from typing import Protocol
from order_service.core.order.domain import RegisterCommand, RegisterResult, PaymentId


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

Адаптер в adapters/out/sber/:

# adapters/out/sber/sber_client_adapter.py
import httpx
from order_service.core.order.domain import RegisterCommand, RegisterResult, PaymentId
from order_service.core.order.port.out.payment_port import PaymentPort
from .sber_mapper import SberMapper
from .exceptions import SberError


class SberClientAdapter:
    def __init__(self, client: httpx.AsyncClient, mapper: SberMapper) -> None:
        self._client = client
        self._mapper = mapper

    async def register(self, cmd: RegisterCommand) -> RegisterResult:
        payload = self._mapper.to_api(cmd)
        try:
            response = await self._client.post("/payment/rest/register.do", json=payload)
            response.raise_for_status()
        except httpx.HTTPError as exc:
            raise SberError(f"Sber register failed: {exc}") from exc
        return self._mapper.to_domain(response.json())

    async def cancel(self, payment_id: PaymentId) -> None:
        try:
            response = await self._client.post(
                "/payment/rest/reverse.do",
                json={"orderId": str(payment_id.value)},
            )
            response.raise_for_status()
        except httpx.HTTPError as exc:
            raise SberError(f"Sber cancel failed: {exc}") from exc

Python не имеет compile-time проверки соответствия Protocol — SberClientAdapter не объявляет явный implements. Структурная совместимость проверяется в тестах:

# tests/adapters/out/sber/test_sber_client_adapter.py
from order_service.core.order.port.out.payment_port import PaymentPort
from order_service.adapters.out.sber.sber_client_adapter import SberClientAdapter


def test_sber_adapter_satisfies_protocol() -> None:
    assert issubclass(SberClientAdapter, PaymentPort)  # type: ignore[misc]

issubclass работает для структурных Protocol — Python проверит наличие и сигнатуры методов.

DI-связывание порт → адаптер происходит в app/container.py:

# app/container.py
from order_service.adapters.out.sber.sber_client_adapter import SberClientAdapter
from order_service.adapters.out.sber.sber_mapper import SberMapper

def build_payment_port(client: httpx.AsyncClient) -> PaymentPort:
    return SberClientAdapter(client=client, mapper=SberMapper())

Handler в core/ инжектирует PaymentPort — конкретный класс ему неизвестен.

Маппер в пакете адаптера

R-HEX-AOUT-3: отдельный модуль <system>_mapper.py переводит между domain и DTO внешней системы.

# adapters/out/sber/sber_mapper.py
from decimal import Decimal
from order_service.core.order.domain import (
    RegisterCommand, RegisterResult, PaymentId, PaymentStatus, Money,
)


_SBER_STATUS_MAP: dict[int, PaymentStatus] = {
    0: PaymentStatus.REGISTERED,
    1: PaymentStatus.AUTHORIZED,
    2: PaymentStatus.DEPOSITED,
    3: PaymentStatus.CANCELLED,
}


class SberMapper:
    def to_api(self, cmd: RegisterCommand) -> dict:
        return {
            "orderNumber": str(cmd.order_id.value),
            "amount": int(cmd.amount.amount * 100),  # Sber принимает копейки
            "currency": 978,                          # 978 = RUB по ISO 4217
            "description": cmd.description,
            "returnUrl": str(cmd.return_url),
        }

    def to_domain(self, data: dict) -> RegisterResult:
        raw_status = data.get("orderStatus", 0)
        status = _SBER_STATUS_MAP.get(raw_status)
        if status is None:
            raise SberError(f"Unknown Sber status: {raw_status}")
        return RegisterResult(
            payment_id=PaymentId(value=data["orderId"]),
            form_url=data["formUrl"],
            status=status,
        )

Что важно:

Маппер знает Sber-специфику. Копейки вместо рублей, числовые коды валют, числовые коды статусов — всё это деталь sber/, не утекает в core/.
Двусторонний. to_api(cmd) для запроса, to_domain(data) для ответа.
Маппер — plain Python. Никакой магии: простые dict, Pydantic-модели Sber (если есть SDK), dataclass. Если внешняя система даёт Pydantic-схему — используй её внутри адаптера, в core/ она не попадает.

Пример с persistence-адаптером (SQLAlchemy):

# adapters/out/persistence/order_mapper.py
from order_service.core.order.domain import Order, OrderId, OrderStatus, Money
from .models import OrderRow  # SQLAlchemy ORM-модель, не domain-тип


class OrderMapper:
    def to_row(self, order: Order) -> OrderRow:
        return OrderRow(
            id=str(order.id.value),
            customer_id=str(order.customer_id.value),
            status=order.status.value,
            total_amount=order.total.amount,
            total_currency=order.total.currency,
        )

    def to_domain(self, row: OrderRow) -> Order:
        return Order(
            id=OrderId(value=row.id),
            customer_id=CustomerId(value=row.customer_id),
            status=OrderStatus(row.status),
            total=Money(amount=row.total_amount, currency=row.total_currency),
        )

OrderRow — ORM-модель SQLAlchemy, она никогда не уходит за пределы adapters/out/persistence/.

Что адаптер знает и не знает

R-HEX-AOUT-4:

adapters/out/persistence/ знает SQLAlchemy, asyncpg, alembic. Не знает про Sber, Kafka.
adapters/out/sber/ знает httpx, Sber-DTO, tenacity. Не знает про PG, Kafka.
adapters/out/kafka/ знает aiokafka, схему сериализации. Не знает про Sber, PG.

import-linter с типом layers гарантирует, что adapters/out/sber/ не импортирует adapters/out/persistence/ и наоборот. Дополнительный контракт independence (см. выше) фиксирует это явно.

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

Антипаттерн	Правило	Что взамен
Port-метод возвращает `SberRegisterResponse` (DTO внешней системы)	`R-HEX-AOUT-X1`	Маппер конвертирует в domain `RegisterResult` внутри адаптера
`if sber_data["orderStatus"] == 4: send_notification(...)` в адаптере	`R-HEX-AOUT-X2`	Решение о последствиях — в handler'е `core/`; адаптер только мапит
Один класс `UniversalAdapter` реализует `PaymentPort`, `SmsPort`, `StoragePort`	`R-HEX-AOUT-X3`	Отдельный класс на каждую систему и порт; per-system isolation
`SberClientAdapter.__init__` принимает `PersistenceAdapter`	`R-HEX-AOUT-X4`	Координация двух адаптеров — use case в `core/`, handler инжектирует оба порта

Бизнес-логика в адаптере

# ПЛОХО — adapters/out/sber/sber_client_adapter.py
async def register(self, cmd: RegisterCommand) -> RegisterResult:
    if cmd.amount.amount > Decimal("100000"):          # ← бизнес-правило
        raise PaymentTooLargeError(cmd.amount)
    data = await self._call_sber(cmd)
    if data.get("orderStatus") == 4:                   # ← интерпретация
        await self._notify_customer(cmd.order_id)      # ← side-effect через другой port
    return self._mapper.to_domain(data)

Что не так:

Правило «100 000 — лимит» живёт в Order.create(...) или handler'е. Если завтра появится OdnaKassa, логику придётся дублировать.
_notify_customer — это другой порт (NotificationPort), его дёргает handler. Адаптер дёргать другой адаптер — нарушение симметрии.
Тесты адаптера должны проверять только маппинг и сетевые сбои, не «что произойдёт после ответа Sber'а».

# ХОРОШО — adapters/out/sber/sber_client_adapter.py
async def register(self, cmd: RegisterCommand) -> RegisterResult:
    payload = self._mapper.to_api(cmd)
    try:
        response = await self._client.post("/payment/rest/register.do", json=payload)
        response.raise_for_status()
    except httpx.HTTPError as exc:
        raise SberError(f"Sber register failed: {exc}") from exc
    return self._mapper.to_domain(response.json())

Координация адаптеров в use case

Если бизнес-сценарий требует «попробовать Sber, при отказе — OdnaKassa»:

# ПЛОХО — adapters/out/sber/sber_client_adapter.py
class SberClientAdapter:
    def __init__(self, ok_adapter: OkClientAdapter) -> None:  # ← инжектирует другой adapter
        self._ok = ok_adapter

# ХОРОШО — core/order/usecase/register_payment_handler.py
class RegisterPaymentHandler:
    def __init__(
        self,
        sber_port: SberPaymentPort,
        ok_port: OkPaymentPort,
    ) -> None:
        self._sber = sber_port
        self._ok = ok_port

    async def handle(self, cmd: RegisterPaymentCommand) -> Payment:
        try:
            return await self._sber.register(cmd)
        except PaymentPortError:
            return await self._ok.register(cmd)   # ← логика выбора в core, не в adapter

Handler инжектирует оба порта. app/container.py связывает каждый порт со своим адаптером. Адаптеры не знают друг о друге.

Куда дальше

Adapters in — симметричная сторона: как FastAPI-роутер маппит request-DTO в UseCase command.
Ports — что именно реализует out-адаптер: Protocol, доменные типы в сигнатурах, port-исключения.
Bootstrap / composition root — как app/container.py связывает порты с адаптерами через DI.
Core layer — почему core/ не знает ни SQLAlchemy, ни httpx, ни FastAPI.
Module structure — полная раскладка пакетов и import-linter-контракт.
Architecture tests — как import-linter фиксирует границы адаптеров в CI.
Когда переходить на Hexagonal — признаки, что пора вводить полную раскладку.