Конфигурация Resilience на Python — pydantic-settings, не хардкод

Опирается на правила: R-RES-CFG-1 … R-RES-CFG-3 и R-RES-CFG-X1 из Resilience Rules → раздел 8. Конфигурация.

Важно знать

• Все параметры CB, retry, bulkhead и timeout — в pydantic-settings, не константами в коде.
• Модель строится как <System>ClientSettings с секцией client.<system>__* в переменных окружения.
• Defaults задаются значениями по умолчанию в модели; per-system — отдельной вложенной моделью или override-классом.
• Имена инстансов = имя системы: sber, insurance, receipt — одно имя для клиента, CB и семафора (R-RES-ISO-3).
• pydantic-settings читает переменные окружения, .env-файлы и секреты из Vault — параметры меняются без redeploy.
• Иерархия timeout (connect < read < total) описывается в настройках явно, программная сборка httpx.Timeout — из модели.
• Программный хардкод числовых литералов в коде адаптера — нарушение R-RES-CFG-X1.

Параметры resilience — это операционные регулировки: failure_rate_threshold, max_concurrent, read_timeout. Их подкручивают на основе наблюдаемого поведения в проде. Если они вписаны константами в конструктор адаптера, любое изменение требует пересборки и выкатки. Если в pydantic-settings — меняются переменной окружения или секретом Vault, pod перезапускается с новым конфигом или получает его через live-reload. Раскрытие раздела 8 гайда.

Модель настроек per-system

R-RES-CFG-1: параметры — в pydantic-settings, не в коде.

# adapters/out/sber/sber_settings.py
from pydantic import BaseModel
from pydantic_settings import BaseSettings, SettingsConfigDict


class TimeoutSettings(BaseModel):
    connect: float = 2.0
    read: float = 30.0
    write: float = 10.0
    pool: float = 5.0
    total: float = 35.0


class CircuitBreakerSettings(BaseModel):
    window_size: int = 50
    min_calls: int = 10
    failure_rate: float = 50.0
    open_duration: float = 30.0
    half_open_calls: int = 3
    slow_call_threshold: float = 15.0


class RetrySettings(BaseModel):
    max_attempts: int = 3
    wait_initial: float = 0.5
    wait_multiplier: float = 2.0
    wait_max: float = 10.0


class BulkheadSettings(BaseModel):
    max_concurrent: int = 10
    acquire_timeout: float = 0.1


class SberClientSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="CLIENT_SBER__")

    timeout: TimeoutSettings = TimeoutSettings(
        connect=2.0,
        read=30.0,
        total=35.0,
        slow_call_threshold=15.0,  # CB slow_call = read / 2
    )
    circuit_breaker: CircuitBreakerSettings = CircuitBreakerSettings(
        failure_rate=30.0,         # платёжная система — порог ниже (R-RES-CB-3)
    )
    retry: RetrySettings = RetrySettings()
    bulkhead: BulkheadSettings = BulkheadSettings(max_concurrent=16)

Переменные окружения с вложенными моделями — через __ (double-underscore):

CLIENT_SBER__CIRCUIT_BREAKER__FAILURE_RATE=20
CLIENT_SBER__TIMEOUT__READ=45
CLIENT_SBER__BULKHEAD__MAX_CONCURRENT=24

Defaults и per-system переопределения

R-RES-CFG-2: общие defaults — в базовой модели, per-system — только то, что реально отличается.

# adapters/out/_base_settings.py
class BaseClientSettings(BaseSettings):
    timeout: TimeoutSettings = TimeoutSettings()
    circuit_breaker: CircuitBreakerSettings = CircuitBreakerSettings()
    retry: RetrySettings = RetrySettings()
    bulkhead: BulkheadSettings = BulkheadSettings()


# adapters/out/sber/sber_settings.py
class SberClientSettings(BaseClientSettings):
    model_config = SettingsConfigDict(env_prefix="CLIENT_SBER__")

    circuit_breaker: CircuitBreakerSettings = CircuitBreakerSettings(
        failure_rate=30.0,   # ← только отклонение от стандарта: платёж, порог ниже
        slow_call_threshold=15.0,
    )
    bulkhead: BulkheadSettings = BulkheadSettings(max_concurrent=16)


# adapters/out/insurance/insurance_settings.py
class InsuranceClientSettings(BaseClientSettings):
    model_config = SettingsConfigDict(env_prefix="CLIENT_INSURANCE__")
    # всё стандартное — не нужно ничего переопределять

При появлении новой системы — новый класс с env_prefix и одним model_config. Системы с нестандартными параметрами видны как отклонения — сигнал задокументировать причину.

Имена инстансов — same as system

R-RES-CFG-3: имя семафора, CB-инстанса и клиента совпадают с именем системы.

# adapters/out/sber/sber_adapter.py
SYSTEM_NAME = "sber"

class SberAdapter:
    def __init__(self, settings: SberClientSettings) -> None:
        self._settings = settings
        self._client = build_client(SYSTEM_NAME, settings)   # httpx.AsyncClient
        self._breaker = build_circuit_breaker(SYSTEM_NAME, settings.circuit_breaker)
        self._sem = asyncio.Semaphore(settings.bulkhead.max_concurrent)

    async def register(self, order: Order) -> PaymentRef:
        async with self._sem:
            try:
                async with self._breaker:
                    async with asyncio.timeout(self._settings.timeout.total):
                        resp = await self._client.post(
                            "/register", json=to_sber_dto(order)
                        )
                return to_payment_ref(resp.json())
            except CircuitBreakerError as e:
                raise PaymentPortError.system_unavailable(SYSTEM_NAME) from e

Имена — короткие, нижний регистр, без хост-частей: sber, не sber_payment_prod_eu_west_1.

Сборка httpx.AsyncClient из настроек

Клиент строится из настроек — числовых литералов в httpx.AsyncClient(...) нет:

# adapters/out/_factory.py
import httpx


def build_client(system: str, settings: BaseClientSettings) -> httpx.AsyncClient:
    t = settings.timeout
    return httpx.AsyncClient(
        timeout=httpx.Timeout(
            connect=t.connect,
            read=t.read,
            write=t.write,
            pool=t.pool,
        ),
        limits=httpx.Limits(
            max_connections=int(settings.bulkhead.max_concurrent * 1.2),
            max_keepalive_connections=settings.bulkhead.max_concurrent,
        ),
    )

Иерархия connect < read < total соблюдается через дефолты в TimeoutSettings; если read переопределяется через env, total должен обновляться вместе — это вариант добавить @model_validator(mode='after'):

class TimeoutSettings(BaseModel):
    connect: float = 2.0
    read: float = 30.0
    write: float = 10.0
    pool: float = 5.0
    total: float = 0.0       # 0 = auto

    @model_validator(mode="after")
    def _fix_total(self) -> "TimeoutSettings":
        if self.total == 0.0:
            self.total = self.read + 5.0
        return self

Регистрация в DI-контейнере

Настройки сначала, клиенты и адаптеры — из них:

# app/dependencies.py
from functools import lru_cache
from adapters.out.sber.sber_settings import SberClientSettings
from adapters.out.insurance.insurance_settings import InsuranceClientSettings


@lru_cache
def get_sber_settings() -> SberClientSettings:
    return SberClientSettings()


@lru_cache
def get_insurance_settings() -> InsuranceClientSettings:
    return InsuranceClientSettings()


async def get_sber_adapter(
    settings: SberClientSettings = Depends(get_sber_settings),
) -> SberAdapter:
    return SberAdapter(settings)

В FastAPI роутере — Depends(get_sber_adapter). Никаких числовых литералов снаружи модели.

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

Куда дальше

• Async и polling — когда asyncio.sleep допустим, а когда — task-queue.
• Bulkhead — sizing asyncio.Semaphore, acquire_timeout, связь с pool.
• Circuit Breaker — параметры count-based окна и slow-call threshold.
• Fallback — когда допустим, контракт except CircuitBreakerError.
• Health checks — FastAPI health с TTL-кешем на каждую систему.
• Observability — prometheus-client метрики CB, retry, semaphore.
• OpenAPI generator binding — CB/retry на адаптере, не на сгенерированном клиенте.
• Per-system isolation — отдельный AsyncClient на каждую внешнюю систему.
• Retry — tenacity, exponential backoff, идемпотентность.
• Timeouts — httpx.Timeout + asyncio.timeout, иерархия connect < read < total.
• Where protection goes — что защищается и где (outbound, inbound, schedulers).