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

Ваш сервис должен сохранить заказ в базу, отправить платёж в банк и положить файл в S3. Это три разных внешних системы, каждая со своим API, SDK и поведением при сбое. Если всё это смешать в одном месте — получится код, который сложно тестировать и ещё сложнее менять. Out-адаптеры решают именно эту проблему.

Что такое out-адаптер

В гексагональной архитектуре ядро приложения (бизнес-логика) не знает ничего про базы данных, HTTP-клиенты или брокеры сообщений. Вместо этого оно работает с портами — абстрактными интерфейсами вроде «сохрани заказ» или «проведи платёж».

Out-адаптер — это конкретная реализация такого порта для одной внешней системы. Он принимает доменный вызов, переводит его в формат внешней системы, делает запрос и возвращает результат обратно в доменных типах.

core/ (бизнес-логика)
  └── вызывает PaymentPort.register(cmd)
        ↓
adapters/out/sber/ (out-адаптер)
  └── маппит cmd → Sber JSON
  └── делает POST /payment/rest/register.do
  └── маппит ответ → RegisterResult
        ↓
возвращает RegisterResult в core/

На этом ответственность адаптера заканчивается. Он не решает, что делать с результатом — это дело бизнес-логики в core/.

Один пакет на одну внешнюю систему

Первая привычка, которую стоит выработать: каждая внешняя система живёт в своём отдельном пакете adapters/out/<system>/.

src/order_service/
  adapters/out/
    persistence/        # SQLAlchemy + PostgreSQL
    sber/               # платёжный шлюз Sбер
    ok/                 # альтернативный платёжный шлюз
    sms/                # SMS-уведомления
    kafka/              # публикация событий
    s3/                 # хранение файлов

Почему это важно:

  • Замена провайдера не ломает остальных. Переехали с одного SMS-провайдера на другой — правите только adapters/out/sms/. Код оплаты и базы данных не трогается.
  • Настройки отказоустойчивости — отдельно для каждого. Тайм-аут для Sбера и тайм-аут для S3 — разные числа с разной логикой повтора. Общий HTTP-клиент на все внешние вызовы — это единая точка отказа.
  • Метрики не смешиваются. payment_sber_errors_total и sms_smsc_errors_total — разные счётчики. Понять, где именно что сломалось, гораздо проще.
  • Тесты изолированы. Mock для Sбера поднимается только в тестах adapters/out/sber/ и не мешает тестам SMS-адаптера.

Чтобы случайно не нарушить изоляцию, можно зафиксировать её через 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",
]

Если один адаптер случайно заимпортирует другой — CI сразу упадёт с понятной ошибкой.

Порт и адаптер: контракт и реализация

Порт — это Protocol в core/. Он описывает только то, что нужно бизнес-логике, в доменных терминах:

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


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

Адаптер реализует этот контракт, не объявляя об этом явно — Python использует структурную совместимость. Достаточно, чтобы у класса были нужные методы с правильными сигнатурами:

# adapters/out/sber/sber_client_adapter.py
import httpx
from order_service.core.order.domain import RegisterCommand, RegisterResult, PaymentId
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 не проверяет соответствие Protocol во время компиляции, соответствие фиксируют тестом:

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

Это работает, потому что PaymentPort помечен @runtime_checkable. Python проверит наличие методов. Полную проверку сигнатур делает mypy статически.

Связывание порта с адаптером происходит в точке сборки приложения:

# app/container.py
def build_payment_port(client: httpx.AsyncClient) -> PaymentPort:
    return SberClientAdapter(client=client, mapper=SberMapper())

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

Маппер: перевод между двумя мирами

Внешние системы используют свои форматы: копейки вместо рублей, числовые коды статусов, специфичные поля. Всё это нельзя пропускать в core/ — там должны быть только доменные типы.

Поэтому в каждом пакете адаптера есть отдельный модуль-маппер:

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


_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),  # Sбер принимает копейки
            "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/, они не утекают в core/.
  • Двусторонний. to_api — для запроса, to_domain — для ответа.
  • Простой код. Обычные dict, датаклассы, Pydantic-модели из SDK системы. Если внешняя система даёт Pydantic-схему — используйте её внутри адаптера, в core/ она не попадает.

То же самое работает для адаптера к базе данных:

# 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/.

Частая ошибка: бизнес-логика в адаптере

Адаптер отвечает за одно: перевести вызов туда и обратно. Как только в нём появляется условная логика по поводу результата — это сигнал, что что-то пошло не так.

# Плохо: адаптер принимает решения
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)    # ← побочный эффект
    return self._mapper.to_domain(data)

Проблемы:

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

Правильно:

# Хорошо: адаптер только маппит и вызывает
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())

Частая ошибка: адаптер внутри адаптера

Иногда хочется сделать так: «если Сбер не отвечает, попробуем через другой шлюз». И соблазн — положить эту логику прямо в SberClientAdapter, передав туда OkClientAdapter.

# Плохо: адаптер знает о другом адаптере
class SberClientAdapter:
    def __init__(self, ok_adapter: OkClientAdapter) -> None:
        self._ok = ok_adapter

Это нарушает изоляцию. Теперь sber/ зависит от ok/, и import-linter это поймает.

Решение — вынести логику переключения в handler:

# Хорошо: координация в core/
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

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

Коротко

  • Out-адаптер — реализация порта для одной внешней системы: маппит доменный вызов → запрос, делает вызов, маппит ответ → доменный результат.
  • Каждая внешняя система — отдельный пакет adapters/out/<system>/. Замена провайдера не затрагивает остальные адаптеры.
  • import-linter с контрактом independence гарантирует, что адаптеры не импортируют друг друга.
  • Адаптер реализует Protocol из core/ структурно (без явного implements). Соответствие проверяется тестом через issubclass и mypy.
  • Маппер — отдельный модуль в пакете адаптера. Он знает специфику внешней системы; в core/ эта специфика не попадает.
  • Бизнес-логика в адаптере — ошибка: правила живут в core/, адаптер только транслирует вызовы.
  • Координация двух адаптеров (fallback, retry между системами) — в handler'е, не внутри адаптера.

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

  • Adapters in — симметричная сторона: как FastAPI-роутер маппит входящий запрос в команду use case.
  • Ports — что именно реализует out-адаптер: Protocol, доменные типы в сигнатурах, исключения порта.
  • Core layer — почему core/ не знает ни SQLAlchemy, ни httpx, ни FastAPI.
  • Bootstrap / composition root — как app/container.py связывает порты с адаптерами.