API 调用错误处理实战：分层定位、有纪律地重试与可观测性设计

# 1. 验证域名解析是否正常
dig +short api.example.com

# 2. 验证 TCP/TLS 握手与证书链是否完整
openssl s_client -connect api.example.com:443 -servername api.example.com < /dev/null 2>&1 | head -20

# 3. 绕开应用代码直接发起请求，观察完整交互过程
curl -v --connect-timeout 5 https://api.example.com/health

"""生产级 API 客户端封装：超时控制、语义化异常与有纪律的重试。"""

import logging
import random
import time
from dataclasses import dataclass, field

import httpx

logger = logging.getLogger("api_client")


class ApiError(Exception):
    """API 调用错误基类，携带状态码与响应体，便于上层按语义处理。"""

    def __init__(self, status_code: int, message: str, body: str = ""):
        self.status_code = status_code
        self.body = body
        super().__init__(f"[{status_code}] {message}")


class RateLimitError(ApiError):
    """429 专用异常：上层捕获后应触发限流退避。"""


@dataclass(frozen=True)
class RetryPolicy:
    """重试策略配置。

    Attributes:
        max_attempts: 最大尝试次数（含首次请求）。
        base_delay: 首次退避的基础间隔（秒）。
        max_delay: 退避间隔上限（秒），防止长时间阻塞。
    """

    max_attempts: int = 3
    base_delay: float = 1.0
    max_delay: float = 30.0


class ApiClient:
    """带分离式超时、语义化异常与指数退避重试的 HTTP 客户端。

    重试决策完全由错误分类学驱动：
    - 瞬态错误（连接失败、读超时、429、5xx）→ 指数退避后重试
    - 持久性错误（400/401/403/404 等）→ 立即抛出，不做无意义重试
    """

    def __init__(
        self,
        base_url: str,
        timeout: float = 10.0,
        retry_policy: RetryPolicy = field(default_factory=RetryPolicy),
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.policy = retry_policy
        # 连接超时与读超时分离：握手失败与响应缓慢是两类不同问题
        self._client = httpx.Client(
            timeout=httpx.Timeout(timeout, connect=5.0),
            headers={"Accept": "application/json"},
        )

    def get_json(self, path: str, **params: str) -> dict:
        """发起 GET 请求并返回解析后的 JSON。

        Args:
            path: 接口路径，如 "/v1/items"。
            **params: 附加到 query string 的参数。

        Returns:
            服务端返回的 JSON 反序列化结果。

        Raises:
            ApiError: 持久性错误，或重试预算耗尽后的最终失败。
            RateLimitError: 429 且重试预算耗尽。
        """
        url = f"{self.base_url}{path}"
        last_error: Exception | None = None

        for attempt in range(1, self.policy.max_attempts + 1):
            try:
                resp = self._client.get(url, params=params)
                if resp.status_code == 429:
                    # 优先遵循服务端的 Retry-After 指示
                    retry_after = float(resp.headers.get("Retry-After", 0))
                    last_error = RateLimitError(
                        429, "rate limited", resp.text
                    )
                    self._wait(retry_after or self._delay(attempt))
                    continue
                if resp.status_code >= 500:
                    last_error = ApiError(
                        resp.status_code, "server error", resp.text
                    )
                    self._wait(self._delay(attempt))
                    continue
                if resp.status_code >= 400:
                    # 持久性错误：重试无意义，立即失败
                    raise ApiError(resp.status_code, "client error", resp.text)
                return resp.json()

            except httpx.TransportError as exc:
                # 传输层异常：连接失败、TLS 握手失败、读超时等，属瞬态
                last_error = exc
                self._wait(self._delay(attempt))

        assert last_error is not None
        raise last_error

    def _delay(self, attempt: int) -> float:
        """全抖动指数退避（full jitter）：interval = random(0, min(2^n, max))。

        加入随机抖动是为了避免惊群效应——大量客户端在相同时间点
        集体重试，会对刚恢复的服务端造成二次冲击。
        """
        ceiling = min(self.policy.base_delay * (2 ** attempt), self.policy.max_delay)
        return random.uniform(0, ceiling)

    def _wait(self, seconds: float) -> None:
        logger.info("retrying in %.1fs", seconds)
        time.sleep(seconds)

proxy_client = httpx.Client(
    proxy="http://用户名:密码@tunnel.16yun.cn:端口",
    timeout=httpx.Timeout(15.0, connect=5.0),
)

import logging
import time
import uuid
logger = logging.getLogger("api_client")
def log_failure(resp: httpx.Response, context: dict) -> str:
 """以结构化格式记录失败请求的完整现场。
    Returns:
        本次故障的唯一 correlation_id，用于向服务商提交工单时引用。
    """
 correlation_id = uuid.uuid4().hex[:12]
 logger.error(
 "api_failed | cid=%s | url=%s | status=%s | elapsed=%.2fs | body=%s | ctx=%s",
 correlation_id,
 resp.request.url,
 resp.status_code,
 resp.elapsed.total_seconds(),
 resp.text[:500], # 响应体截断，防止日志膨胀与敏感信息全量落盘
 context,
 )
 return correlation_id