python-resilience

来源声明：本技能内容源自 wshobson/agents （MIT License，Copyright (c) 2024 Seth Hobson；完整许可文本见同目录 LICENSE.txt）。 使用、修改和再分发须遵守该 MIT 许可条款并保留版权声明。

中文描述

本技能提供全栈开发相关的最佳实践和模式指导。

---

name: python-resilience description: 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:
    """Fetch data with automatic retry on transient failures."""
    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

# Define what's retryable
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:
    """Make API call with retry on network issues."""
    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:
    """Check if response indicates a retryable error."""
    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:
    """Make HTTP request with retry on transient status codes."""
    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 call with comprehensive retry handling."""
    return httpx.request(method, url, timeout=30, **kwargs)

详细示例和模式

详细内容（从 ## Advanced Patterns 开始）位于 references/details.md。如果上方的导航摘要不足以满足需求，请阅读该文件。

最佳实践摘要

1. 仅重试暂时性错误——不要重试程序缺陷或身份验证失败
2. 使用指数退避——给服务留出恢复时间
3. 加入随机抖动——避免同步重试引发惊群效应
4. 限制总时长——stop_after_attempt(5) | stop_after_delay(60)
5. 记录每次重试——静默重试会掩盖系统性问题
6. 使用装饰器——将重试逻辑与业务逻辑分离
7. 注入依赖项——让基础设施易于测试
8. 为所有操作设置超时——每次网络调用都需要超时
9. 优雅失败——对非关键路径返回缓存值或默认值
10. 监控重试率——高重试率表明存在潜在问题

能力边界

✅ 擅长处理

• 全栈开发相关的最佳实践
• 代码架构设计指导
• 性能优化建议

⚠️ 需要素材

• 具体的项目上下文
• 技术栈信息

❌ 超出范围

• 非技术领域的业务决策
• 具体的代码实现（仅提供指导）

工作流程

步骤 1：需求分析

理解项目需求和技术约束

步骤 2：架构设计

设计系统架构和组件划分

步骤 3：实现指导

提供具体的实现建议和代码模式

步骤 4：代码审查

审查代码质量和最佳实践遵循情况

步骤 5：优化建议

提供性能优化和改进建议

注意事项

1. 过度设计

避免在简单场景中使用复杂的设计模式

2. 忽视性能

不要忽视性能优化，特别是在生产环境中

3. 安全考虑

始终考虑安全性和数据保护

4. 测试覆盖

确保有足够的测试覆盖，特别是边界情况

5. 文档维护

保持文档与代码同步更新