Опирается на правила: R-RES-OAS-1 … R-RES-OAS-4 и R-RES-OAS-X1 … R-RES-OAS-X3 из Resilience Rules → раздел 9. Связка с OpenAPI generator.

Важно знать

  • CB / retry / bulkhead — на public-методе out-adapter класса, который оборачивает вызов сгенерированного клиента. Не на сгенерированном клиенте, не в helper-функции.
  • Сгенерированный клиент перегенерируется из YAML при каждой сборке — любые правки в нём пропадут.
  • Инструмент генерации — openapi-python-client (httpx-native) или openapi-generator с python-target; спека внешнего API в adapters/out/<system>/openapi/<system>.yaml.
  • Сгенерированный код не коммитится; добавляется в .gitignore.
  • Mapper обязателен между сгенерированными DTO и domain-типами из core/. Сгенерированные DTO — транспорт, не доменные типы.
  • Port-метод возвращает domain-тип (PaymentRef, OrderStatus), а не DTO внешней системы.
  • Все resilience-параметры — через pydantic-settings (<System>ClientSettings), не хардкодом.

OpenAPI-first для outbound: внешний API описан YAML-ом, из него генерируется httpx-клиент, вокруг него пишется адаптер. Ключевой вопрос — где именно крепить защиту, и почему именно там.

Обёртки на public-методе out-adapter

R-RES-OAS-1: asyncio.Semaphore (bulkhead), CB (purgatory/aiobreaker), tenacity (retry) — на нашем adapter-методе, оборачивающем сгенерированный клиент.

# adapters/out/sber/generated/  ← не модифицируем (перегенерируется)
# openapi-python-client создаёт: SberClient, RegisterRequest, RegisterResponse, GetOrderStatusResponse...

# core/payment/port/out/payment_port.py — domain port:
from abc import ABC, abstractmethod
from core.payment.domain import Order, PaymentRef, OrderStatus, OrderId

class PaymentPort(ABC):
    @abstractmethod
    async def register(self, order: Order) -> PaymentRef: ...

    @abstractmethod
    async def get_status(self, order_id: OrderId) -> OrderStatus: ...


# adapters/out/sber/sber_adapter.py — наш adapter:
import asyncio
import tenacity
from purgatory import CircuitBreakerError

from adapters.out.sber.generated import SberClient
from adapters.out.sber.mapper import to_sber_request, to_domain_ref, to_domain_status
from adapters.out.sber.settings import SberClientSettings
from core.payment.domain import Order, PaymentRef, OrderStatus, OrderId
from core.payment.port.out.payment_port import PaymentPort
from core.payment.exception import PaymentSystemUnavailable


class SberAdapter(PaymentPort):
    def __init__(
        self,
        client: SberClient,
        breaker,
        semaphore: asyncio.Semaphore,
        settings: SberClientSettings,
    ) -> None:
        self._client = client
        self._breaker = breaker
        self._sem = semaphore
        self._settings = settings

    async def register(self, order: Order) -> PaymentRef:
        async with self._sem:
            try:
                async with self._breaker:
                    async with asyncio.timeout(self._settings.total_timeout):
                        req = to_sber_request(order)
                        resp = await self._client.register.asyncio(body=req)
                        return to_domain_ref(resp)
            except CircuitBreakerError as exc:
                raise PaymentSystemUnavailable("sber") from exc

    @tenacity.retry(
        wait=tenacity.wait_exponential(multiplier=0.5, min=0.5, max=4),
        stop=tenacity.stop_after_attempt(3),
        retry=tenacity.retry_if_exception_type((TimeoutError, httpx.ConnectError)),
        reraise=True,
    )
    async def get_status(self, order_id: OrderId) -> OrderStatus:
        async with self._sem:
            async with self._breaker:
                async with asyncio.timeout(self._settings.total_timeout):
                    resp = await self._client.get_order_status.asyncio(
                        order_id=order_id.value
                    )
                    return to_domain_status(resp)

Почему именно здесь:

  • Сгенерированный SberClient перегенерируется при каждом make generate или CI-шаге — любые декораторы/обёртки на его методах исчезнут.
  • Helper-функция с именем системы строкой (execute_with_cb("sber", ...)) теряет compile-time проверку: опечатка в строке обнаружится только в рантайме.
  • SberAdapter.register — публичная граница port PaymentPort. Именно на ней имеет смысл мерить «вызов к Sber» как бизнес-операцию.

Генерация клиента из OpenAPI-спеки

R-RES-OAS-2: для нового кода клиент генерируется из OpenAPI-спеки внешней системы.

openapi-python-client создаёт httpx-native async-клиент, не требующий дополнительной обвязки:

# pyproject.toml — dev-зависимость
[tool.poetry.dev-dependencies]
openapi-python-client = ">=0.21"

# Makefile
generate-sber:
    openapi-python-client generate \
        --path adapters/out/sber/openapi/sber.yaml \
        --output adapters/out/sber/generated \
        --overwrite

# adapters/out/sber/openapi/sber.yaml  ← спека внешней системы, коммитится
# adapters/out/sber/generated/         ← в .gitignore, регенерируется

Что даёт генерация:

  • Типизированный async-клиент на httpx — методы возвращают Pydantic-модели, не dict.
  • Pydantic-валидация ответа внешней системы на входе, до маппинга в domain. Breaking change в API внешней системы ловится на этапе валидации, не на обращении к полю.
  • Обновление спеки через PR — diff в YAML видно в review; сгенерированный клиент не коммитится, и diff его никого не беспокоит.

Альтернатива — openapi-generator с python-target (тоже httpx-native). Выбор инструмента — по команде; принципы одинаковы.

Расположение спеки и codegen

R-RES-OAS-3: OpenAPI-спека внешнего API — в adapters/out/<system>/openapi/<system>.yaml. Сгенерированный код — в .gitignore.

adapters/out/
├── sber/
│   ├── openapi/
│   │   └── sber.yaml              # ← спека внешнего API, коммитится
│   ├── generated/                 # ← в .gitignore, не коммитится
│   │   ├── __init__.py
│   │   ├── client.py
│   │   └── models/
│   │       ├── register_request.py
│   │       └── register_response.py
│   ├── sber_adapter.py            # ← наш адаптер, оборачивает generated
│   ├── mapper.py                  # ← DTO → domain
│   └── settings.py                # ← pydantic-settings
├── receipt/
│   └── ...

Важные детали:

  • sber.yaml — копия (или форк) внешнего OpenAPI. Если внешняя система не публикует спеку — пишется вручную по документации.
  • Версионирование — спека коммитится, generated/ — нет. PR со spec-update показывает YAML-diff, из которого ясно, что изменилось в API.
  • CI — шаг make generate запускается до сборки; если сгенерированный код изменился — сборка падает с подсказкой сделать коммит спеки.

Mapper между сгенерированными DTO и domain

R-RES-OAS-4: между сгенерированным клиентом и port из core/ — обязательный mapper.

# core/payment/domain.py — domain-типы (не зависят от внешней системы):
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum

@dataclass(frozen=True)
class OrderId:
    value: str

@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str

@dataclass(frozen=True)
class PaymentRef:
    external_id: str
    status: "PaymentStatus"

class PaymentStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    DECLINED = "declined"


# adapters/out/sber/mapper.py — DTO → domain:
from adapters.out.sber.generated.models import (
    RegisterRequest,
    RegisterResponse,
    SberStatus,
)
from core.payment.domain import Order, PaymentRef, PaymentStatus, Money


def to_sber_request(order: Order) -> RegisterRequest:
    return RegisterRequest(
        order_id=order.id.value,
        amount=float(order.total.amount),
        currency=order.total.currency,
        idempotency_key=order.idempotency_key,
    )


def to_domain_ref(resp: RegisterResponse) -> PaymentRef:
    return PaymentRef(
        external_id=resp.sber_order_id,
        status=_map_status(resp.status),
    )


def _map_status(status: SberStatus) -> PaymentStatus:
    match status:
        case SberStatus.CONFIRMED:
            return PaymentStatus.CONFIRMED
        case SberStatus.PENDING:
            return PaymentStatus.PENDING
        case SberStatus.DECLINED:
            return PaymentStatus.DECLINED
        case _:
            raise ValueError(f"Unknown Sber status: {status}")

Зачем mapper:

  • Сгенерированные DTO — транспорт. Они меняются вместе с внешним API. Domain — стабильный.
  • Изоляция от breaking changes. Внешняя система переименовала поле — меняется один mapper, остальной код не задет.
  • Явное соответствие статусов. SberStatus.PENDING ≠ PaymentStatus.PENDING семантически — mapper делает explicit-перевод с ветвью case _ на незнакомые значения.

Конфигурация через pydantic-settings

R-RES-CFG-1 / R-RES-OAS-4: параметры клиента — через pydantic-settings, не хардкодом.

# adapters/out/sber/settings.py
from pydantic import Field
from pydantic_settings import BaseSettings


class SberClientSettings(BaseSettings):
    base_url: str = Field(alias="client_sber__base_url")
    connect_timeout: float = Field(default=2.0, alias="client_sber__connect_timeout")
    read_timeout: float = Field(default=10.0, alias="client_sber__read_timeout")
    total_timeout: float = Field(default=12.0, alias="client_sber__total_timeout")
    max_connections: int = Field(default=10, alias="client_sber__max_connections")
    max_keepalive: int = Field(default=8, alias="client_sber__max_keepalive")
    max_concurrent: int = Field(default=8, alias="client_sber__max_concurrent")
    cb_failure_rate: float = Field(default=0.30, alias="client_sber__cb_failure_rate")
    cb_window_size: int = Field(default=50, alias="client_sber__cb_window_size")

    model_config = {"env_nested_delimiter": "__"}

# adapters/out/sber/factory.py — сборка адаптера по настройкам:
import asyncio
import httpx
from purgatory import AsyncCircuitBreaker

from adapters.out.sber.generated import AuthenticatedClient
from adapters.out.sber.sber_adapter import SberAdapter
from adapters.out.sber.settings import SberClientSettings


def build_sber_adapter(settings: SberClientSettings) -> SberAdapter:
    limits = httpx.Limits(
        max_connections=settings.max_connections,
        max_keepalive_connections=settings.max_keepalive,
    )
    timeout = httpx.Timeout(
        connect=settings.connect_timeout,
        read=settings.read_timeout,
        write=settings.read_timeout,
        pool=settings.connect_timeout,
    )
    http_client = httpx.AsyncClient(base_url=settings.base_url, limits=limits, timeout=timeout)
    generated_client = AuthenticatedClient(base_url=settings.base_url, httpx_client=http_client)

    breaker = AsyncCircuitBreaker(
        failure_threshold=settings.cb_failure_rate,
        recovery_timeout=30,
        name="sber",
    )
    semaphore = asyncio.Semaphore(settings.max_concurrent)

    return SberAdapter(generated_client, breaker, semaphore, settings)

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

АнтипаттернПравилоЧто взамен
Декораторы/обёртки на методах сгенерированного клиентаR-RES-OAS-X1На public-методе out-adapter
Helper execute_with_cb("sber", ...) со строковым именемR-RES-OAS-X2CB-инстанс в __init__, применяется на методе адаптера
PaymentPort.register возвращает RegisterResponse (сгенерированный DTO)R-RES-OAS-X3Domain PaymentRef + mapper
Сгенерированный код в gitR-RES-OAS-3.gitignore на adapters/out/<system>/generated/
httpx.AsyncClient() без limits/timeout в сгенерированном клиентеR-RES-ISO-X2Явные httpx.Limits + httpx.Timeout per-system
Один AsyncClient на несколько внешних системR-RES-ISO-X1Отдельный клиент per-system

Куда дальше

  • Per-system isolation — отдельный AsyncClient + CB + семафор на каждую систему.
  • Circuit Breaker — purgatory/aiobreaker, настройка окна и порогов.
  • Bulkhead — asyncio.Semaphore per-system, sizing < pool.
  • Retry — tenacity, exponential backoff, только при идемпотентности.
  • Timeouts — иерархия connect < read < total, asyncio.timeout.
  • Configuration — pydantic-settings, defaults + per-system override.
  • Fallback — допустимые случаи, контракт, запреты для money.
  • Health checks — TTL-кеш probe, FastAPI /health/ready.
  • Observability — prometheus-client, OTel-spans, state-transition лог.
  • Async и polling — task-queue вместо asyncio.sleep-цикла.
  • Где ставить защиту — outbound vs inbound vs schedulers.