Timeouts в Python — httpx.Timeout + asyncio.timeout без бесконечностей

httpx.Timeout(connect/read/write/pool) + asyncio.timeout() как внешний cap. Per-system конфиг через pydantic-settings. TimeBudget из traceparent. Правила R-RES-TO-1..3.

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

Опирается на правила: R-RES-TO-1 … R-RES-TO-3 и R-RES-TO-X1 … R-RES-TO-X3 из Resilience Style Guide → раздел 3. Timeouts.

Важно знать

Иерархия: connect < read < total. Никаких None (∞).

connect — TCP/TLS handshake. Локальные DC: 2s. Через интернет: 5s. Никогда >10s.

read — ожидание байт из сокета. 10s быстрые API, 30s тяжёлые, 60s отчёты. Если >60s — это async-pattern, не sync-вызов.

asyncio.timeout(total) — внешний cap вокруг всего вызова. httpx.Timeout отдельно не даёт единого cap; нужны оба уровня.

Timeouts — per-system через <System>ClientSettings (pydantic-settings, секция client.<system>).

При traceparent с TimeBudget — перехватчик сжимает total до min(total, remaining - 100ms).

httpx.AsyncClient() без timeout= = дефолт httpx.DEFAULT_TIMEOUT_CONFIG (5s везде) или None если явно передан — корень медленного зависания.

Без явных timeouts любой outbound HTTP может зависнуть: TCP-соединение установилось, но байты не приходят. Первые зависшие таски блокируют event loop, последующие накапливаются в очереди, и сервис перестаёт отвечать. Решение — два уровня контроля: httpx.Timeout для детального контроля сокета и asyncio.timeout() как единый cap.

Иерархия двух уровней timeout

R-RES-TO-1: в Python async-стеке timeout задаётся на двух уровнях, которые дополняют друг друга:

Уровень	Механизм	Что измеряет
Сокет	`httpx.Timeout(connect=.., read=.., write=.., pool=..)`	TCP handshake, ожидание байт, запись, взятие из пула
Общий cap	`asyncio.timeout(total)`	Всё время от отправки до полного ответа

Почему нужны оба:

httpx.Timeout контролирует каждую фазу отдельно: можно поймать «медленный TLS-handshake» (connect) и отдельно «сервер застыл после первых байт» (read).
asyncio.timeout() — страховка от редкого сценария: сервер шлёт байт раз в read - 1 секунд бесконечно; read-timeout не срабатывает, но общий cap прерывает вызов.

# adapters/out/sber/sber_client.py
import httpx
import asyncio

t = httpx.Timeout(connect=5.0, read=30.0, write=5.0, pool=1.0)

async with asyncio.timeout(37.0):               # total cap (connect + read + buffer)
    resp = await client.post("/register", json=payload, timeout=t)

На практике timeout= передаётся в AsyncClient при создании — тогда он применяется ко всем запросам этого клиента. asyncio.timeout() оборачивает вызов на уровне адаптерного метода.

Per-system конфиг через pydantic-settings

R-RES-TO-2: timeouts — не magic-numbers в коде, а конфигурация.

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

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

    base_url: str
    connect_timeout: float = Field(default=5.0)
    read_timeout: float = Field(default=30.0)
    write_timeout: float = Field(default=5.0)
    pool_timeout: float = Field(default=1.0)
    total_timeout: float = Field(default=37.0)   # connect + read + 2s buffer
    max_connections: int = Field(default=25)
    max_keepalive_connections: int = Field(default=20)

    def as_httpx_timeout(self) -> httpx.Timeout:
        return httpx.Timeout(
            connect=self.connect_timeout,
            read=self.read_timeout,
            write=self.write_timeout,
            pool=self.pool_timeout,
        )

# config/settings.yml
client:
  sber:
    base_url: "https://api.sber.example.com"
    connect_timeout: 5.0
    read_timeout: 30.0
    total_timeout: 37.0
    max_connections: 25

  odnakassa:
    base_url: "https://api.odnakassa.example.com"
    connect_timeout: 5.0
    read_timeout: 60.0      # OdnaKassa возвращает multiset-ответы
    total_timeout: 67.0
    max_connections: 30

Создание клиента с этими настройками:

# adapters/out/sber/sber_adapter.py
class SberAdapter:
    def __init__(self, settings: SberClientSettings) -> None:
        limits = httpx.Limits(
            max_connections=settings.max_connections,
            max_keepalive_connections=settings.max_keepalive_connections,
        )
        self._client = httpx.AsyncClient(
            base_url=settings.base_url,
            timeout=settings.as_httpx_timeout(),
            limits=limits,
        )
        self._total = settings.total_timeout

    async def register(self, order: Order) -> PaymentRef:
        async with asyncio.timeout(self._total):
            resp = await self._client.post("/register", json=to_sber_dto(order))
        resp.raise_for_status()
        return to_domain(resp.json())

Что важно:

Валидация на старте: pydantic-settings проверяет типы при инициализации приложения; невалидный конфиг даёт ValidationError до того, как сервис принял первый запрос.
Отклонения от типовых — с комментарием в yml (OdnaKassa read_timeout: 60.0 — объяснено).
total_timeout должен быть явным и > read_timeout — иначе asyncio.timeout сработает раньше, чем httpx успеет разобраться в причине.

Уважение TimeBudget при traceparent

R-RES-TO-3: если входящий запрос содержит traceparent + X-Time-Budget header, total-timeout сжимается до оставшегося бюджета.

# adapters/out/_common/time_budget.py
import asyncio
from contextvars import ContextVar
from typing import Optional

_remaining_budget: ContextVar[Optional[float]] = ContextVar("remaining_budget", default=None)

def set_budget(seconds: float) -> None:
    _remaining_budget.set(seconds)

def get_budget() -> Optional[float]:
    return _remaining_budget.get()

# adapters/out/_common/time_budget_middleware.py
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class TimeBudgetMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        budget_header = request.headers.get("x-time-budget")
        if budget_header:
            set_budget(float(budget_header) / 1000.0)   # ms → s
        return await call_next(request)

# adapters/out/sber/sber_adapter.py
async def register(self, order: Order) -> PaymentRef:
    remaining = get_budget()
    if remaining is not None and remaining < 0.1:
        raise PaymentPortError.system_unavailable("sber")

    effective_total = self._total
    if remaining is not None:
        effective_total = min(self._total, remaining - 0.1)

    async with asyncio.timeout(effective_total):
        resp = await self._client.post("/register", json=to_sber_dto(order))
    resp.raise_for_status()
    return to_domain(resp.json())

Зачем: если upstream-вызов сказал «у тебя 2 секунды на ответ», а текущий outbound ждёт 30 секунд — это deadline-violation. upstream уже закрыл соединение, сервис зря держит таску. Сжатие под бюджет даёт fail-fast: лучше отдать 504 сейчас.

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

AsyncClient без timeout

R-RES-TO-X1: httpx.AsyncClient() без явного timeout= использует DEFAULT_TIMEOUT_CONFIG (5s на каждую фазу) — но может быть переопределён на None вызывающей стороной. Явный конфиг устраняет неопределённость.

# ПЛОХО — полагаемся на дефолт, который может быть переопределён
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.sber_client = httpx.AsyncClient(base_url=settings.base_url)
    yield

# ПЛОХО — явный None = ∞
app.state.sber_client = httpx.AsyncClient(base_url=settings.base_url, timeout=None)

Первый зависший response держит таску, следующие накапливаются, event loop деградирует.

asyncio.timeout меньше read

R-RES-TO-X2: asyncio.timeout(total) при total < read_timeout — внутреннее противоречие.

# ПЛОХО — total сработает раньше, чем read успеет что-то поймать
t = httpx.Timeout(connect=5.0, read=30.0)

async with asyncio.timeout(20.0):               # ← меньше read!
    resp = await client.post("/register", ...)

Что не так: asyncio.timeout(20.0) сработает через 20 секунд как asyncio.TimeoutError; httpx-уровень read=30.0 никогда не сработает. Метрики путаются (фиксируется TimeoutError, а не ReadTimeout), error-handling ветвится неожиданно.

Корректно: total = connect + read + buffer(1-2s).

read > 60s для sync-вызова

R-RES-TO-X3: если для синхронного outbound нужен read > 60s — это признак async-pattern, не sync-вызова.

# ПЛОХО — async-handler FastAPI держит таску 5 минут
client:
  reports:
    read_timeout: 300.0     # 5 минут для отчёта
    total_timeout: 310.0

Что не так:

Таска FastAPI заблокирована на 5 минут. При нагрузке исчерпывается пул воркеров uvicorn.
Upstream-клиент (браузер, API Gateway) скорее всего имеет timeout <60s — он уже получил 504, а наша таска ещё висит.
Внешняя система, которой нужно 5 минут на ответ, требует async-pattern: POST /reports → 202 Accepted + task_id, GET /reports/{id} для статуса.

Корректно: task-queue с polling (см. Async и polling).

Что запрещено — таблица

Антипаттерн	Правило	Что взамен
`httpx.AsyncClient()` без явного `timeout=`	`R-RES-TO-X1`	`httpx.Timeout(connect, read, write, pool)` per-system
`asyncio.timeout(total)` при `total < read_timeout`	`R-RES-TO-X2`	`total = connect + read + buffer`
`read_timeout > 60s` для async-handler	`R-RES-TO-X3`	Task-queue + polling / webhook
Magic-numbers timeout в коде	`R-RES-TO-2`	`pydantic-settings` + `<System>ClientSettings`
Игнорирование TimeBudget при `traceparent`	`R-RES-TO-3`	`asyncio.timeout(min(total, remaining - 0.1))`
Один `AsyncClient` на несколько систем	`R-RES-ISO-X1`	Per-system клиент с собственным `httpx.Limits`

Куда дальше

Per-system isolation — отдельный AsyncClient на каждую внешнюю систему.
Circuit Breaker — slow_call_duration_threshold срабатывает раньше timeout.
Bulkhead — asyncio.Semaphore как ограничитель одновременных вызовов.
Async и polling — если нужен outbound >60s.
Конфигурация — pydantic-settings, defaults и per-system override.
Retry — tenacity и согласование с timeout-исключениями.
Observability — метрики timeout через prometheus-client и OTel-спаны.
Health checks — TTL-кеш probe и связь с readiness.
Fallback — обработка asyncio.TimeoutError / httpx.TimeoutException.