skillhub-162-python-resilience/SKILL.md

name, description

name	description
python-resilience	Python 弹性模式，包括自动重试、指数退避、超时和容错装饰器。在需要添加重试逻辑、实现超时、构建容错服务或处理瞬时故障时使用。

Python 弹性模式

构建容错的 Python 应用程序，优雅地处理瞬时故障、网络问题和服务中断。当依赖项不可靠时，弹性模式能让系统保持运行。

何时使用本技能

• 为外部服务调用添加重试逻辑
• 为网络操作实现超时
• 构建容错的微服务
• 处理速率限制和背压
• 创建基础设施装饰器
• 设计断路器

核心概念

1. 瞬时故障与永久故障

重试瞬时错误（网络超时、临时服务问题）。不要重试永久错误（无效凭据、错误请求）。

2. 指数退避

增加重试之间的等待时间，避免让正在恢复的服务不堪重负。

3. 抖动

为退避算法添加随机性，防止多个客户端同时重试时产生惊群效应。

4. 有界重试

同时限制重试次数和总持续时间，防止无限重试循环。

快速开始

from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def call_external_service(request: dict) -> dict:
    return httpx.post("https://api.example.com", json=request).json()

基础模式

模式 1：使用 Tenacity 的基础重试

使用 tenacity 库实现生产级重试逻辑。对于更简单的场景，可考虑内置重试功能或轻量级自定义实现。

from tenacity import (
    retry,
    stop_after_attempt,
    stop_after_delay,
    wait_exponential_jitter,
    retry_if_exception_type,
)

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

@retry(
    retry=retry_if_exception_type(TRANSIENT_ERRORS),
    stop=stop_after_attempt(5) | stop_after_delay(60),
    wait=wait_exponential_jitter(initial=1, max=30),
)
def fetch_data(url: str) -> dict:
    """在瞬时故障时自动重试获取数据。"""
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

模式 2：仅重试适当的错误

白名单特定的瞬时异常。绝不重试：

• ValueError、TypeError——这些是代码缺陷，不是瞬时问题
• AuthenticationError——无效凭据不会自动变有效
• HTTP 4xx 错误（429 除外）——客户端错误是永久性的

from tenacity import retry, retry_if_exception_type
import httpx

# 定义哪些是可重试的
RETRYABLE_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    httpx.ConnectTimeout,
    httpx.ReadTimeout,
)

@retry(
    retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def resilient_api_call(endpoint: str) -> dict:
    """在出现网络问题时带重试的 API 调用。"""
    return httpx.get(endpoint, timeout=10).json()

模式 3：HTTP 状态码重试

重试表示瞬时问题的特定 HTTP 状态码。

from tenacity import retry, retry_if_result, stop_after_attempt
import httpx

RETRY_STATUS_CODES = {429, 502, 503, 504}

def should_retry_response(response: httpx.Response) -> bool:
    """检查响应是否表示可重试的错误。"""
    return response.status_code in RETRY_STATUS_CODES

@retry(
    retry=retry_if_result(should_retry_response),
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def http_request(method: str, url: str, **kwargs) -> httpx.Response:
    """在瞬时状态码上带重试的 HTTP 请求。"""
    return httpx.request(method, url, timeout=30, **kwargs)

模式 4：异常与状态码组合重试

同时处理网络异常和 HTTP 状态码。

from tenacity import (
    retry,
    retry_if_exception_type,
    retry_if_result,
    stop_after_attempt,
    wait_exponential_jitter,
    before_sleep_log,
)
import logging
import httpx

logger = logging.getLogger(__name__)

TRANSIENT_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    httpx.ConnectError,
    httpx.ReadTimeout,
)
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}

def is_retryable_response(response: httpx.Response) -> bool:
    return response.status_code in RETRY_STATUS_CODES

@retry(
    retry=(
        retry_if_exception_type(TRANSIENT_EXCEPTIONS) |
        retry_if_result(is_retryable_response)
    ),
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=30),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def robust_http_call(
    method: str,
    url: str,
    **kwargs,
) -> httpx.Response:
    """带全面重试处理的 HTTP 调用。"""
    return httpx.request(method, url, timeout=30, **kwargs)

详细的完整示例与模式

详细章节（以 ## 高级模式 开头）位于 references/details.md 中。当以上导航摘要不足以满足需求时，请阅读该文件。

最佳实践总结

1. 仅重试瞬时错误——不要重试代码缺陷或认证失败
2. 使用指数退避——给服务恢复的时间
3. 添加抖动——防止同步重试导致惊群效应
4. 限制总时长——stop_after_attempt(5) | stop_after_delay(60)
5. 记录每次重试——静默重试会隐藏系统性问题
6. 使用装饰器——将重试逻辑与业务逻辑分离
7. 注入依赖——使基础设施可测试
8. 处处设置超时——每个网络调用都需要超时
9. 优雅降级——为非关键路径返回缓存/默认值
10. 监控重试率——高重试率表明存在根本性问题