Python httpx ReadTimeout·ConnectError 재시도 설계

import httpx

TIMEOUT = httpx.Timeout(
    connect=2.0,  # DNS+TCP+TLS 포함(환경에 따라 TLS 포함 여부 체감)
    read=10.0,    # 응답 바디 읽기(스트리밍이면 더 길게)
    write=5.0,    # 요청 바디 업로드
    pool=2.0,     # 커넥션 풀에서 대기
)

LIMITS = httpx.Limits(
    max_connections=100,
    max_keepalive_connections=20,
    keepalive_expiry=30.0,
)

client = httpx.Client(timeout=TIMEOUT, limits=LIMITS)

from __future__ import annotations

import random
import time
from dataclasses import dataclass
from typing import Iterable, Optional

import httpx

RETRYABLE_STATUS = {408, 425, 429, 500, 502, 503, 504}
IDEMPOTENT_METHODS = {"GET", "HEAD", "OPTIONS", "PUT", "DELETE"}


@dataclass
class RetryPolicy:
    max_attempts: int = 4
    base_delay: float = 0.2        # seconds
    max_delay: float = 3.0         # seconds
    jitter: float = 0.2            # 0.0~1.0: delay에 곱해질 랜덤 비율

    def backoff(self, attempt: int) -> float:
        # attempt: 1,2,3...
        delay = min(self.max_delay, self.base_delay * (2 ** (attempt - 1)))
        # full jitter 변형: delay 범위 내 랜덤
        return random.uniform(0, delay * (1.0 + self.jitter))


def parse_retry_after(resp: httpx.Response) -> Optional[float]:
    ra = resp.headers.get("Retry-After")
    if not ra:
        return None
    try:
        return float(ra)
    except ValueError:
        return None


def is_idempotent(method: str) -> bool:
    return method.upper() in IDEMPOTENT_METHODS


def should_retry_exception(exc: Exception, method: str) -> bool:
    # 연결 실패는 대체로 재시도 가치가 큼
    if isinstance(exc, httpx.ConnectError):
        return True

    # 읽기 타임아웃은 “이미 서버가 처리했을 수도” 있으므로 멱등 요청만
    if isinstance(exc, httpx.ReadTimeout):
        return is_idempotent(method)

    # 프로토콜 오류/일시적 네트워크 오류도 상황에 따라 재시도 가능
    if isinstance(exc, (httpx.RemoteProtocolError, httpx.PoolTimeout)):
        return is_idempotent(method)

    return False


def should_retry_response(resp: httpx.Response, method: str) -> bool:
    if resp.status_code not in RETRYABLE_STATUS:
        return False
    # 429/503 등은 POST라도 서버가 “다시 시도”를 요구하는 경우가 있어
    # Idempotency-Key를 쓸 수 있으면 POST도 허용하는 편이 현실적.
    if is_idempotent(method):
        return True
    return resp.status_code in {429, 503}


def request_with_retry(
    client: httpx.Client,
    method: str,
    url: str,
    *,
    policy: RetryPolicy = RetryPolicy(),
    idempotency_key: Optional[str] = None,
    **kwargs,
) -> httpx.Response:
    headers = dict(kwargs.pop("headers", {}) or {})
    if idempotency_key:
        headers.setdefault("Idempotency-Key", idempotency_key)

    last_exc: Optional[Exception] = None

    for attempt in range(1, policy.max_attempts + 1):
        try:
            resp = client.request(method, url, headers=headers, **kwargs)

            if should_retry_response(resp, method) and attempt < policy.max_attempts:
                retry_after = parse_retry_after(resp)
                sleep_s = retry_after if retry_after is not None else policy.backoff(attempt)
                resp.close()  # 커넥션 반환
                time.sleep(sleep_s)
                continue

            return resp

        except Exception as exc:
            last_exc = exc
            if attempt >= policy.max_attempts or not should_retry_exception(exc, method):
                raise
            time.sleep(policy.backoff(attempt))

    assert last_exc is not None
    raise last_exc

import uuid
import httpx

client = httpx.Client(timeout=10.0)

key = str(uuid.uuid4())
resp = request_with_retry(
    client,
    "POST",
    "https://api.example.com/payments",
    idempotency_key=key,
    json={"amount": 1000, "currency": "KRW"},
)
resp.raise_for_status()
print(resp.json())

import threading
import httpx

sema = threading.Semaphore(50)  # 외부 API 동시 호출 상한

client = httpx.Client(
    timeout=httpx.Timeout(connect=2, read=10, write=5, pool=2),
    limits=httpx.Limits(max_connections=60, max_keepalive_connections=20),
)

def safe_call(url: str) -> dict:
    with sema:
        resp = request_with_retry(client, "GET", url)
        resp.raise_for_status()
        return resp.json()

import time
import logging
import httpx

log = logging.getLogger("http")

def get_with_metrics(client: httpx.Client, url: str) -> httpx.Response:
    t0 = time.perf_counter()
    try:
        resp = request_with_retry(client, "GET", url)
        dt = time.perf_counter() - t0
        log.info("http_call", extra={"url": url, "status": resp.status_code, "elapsed": dt})
        return resp
    except Exception as e:
        dt = time.perf_counter() - t0
        log.warning("http_call_failed", extra={"url": url, "err": type(e).__name__, "elapsed": dt})
        raise