如何通过tenacity库为不同错误类型实现自定义重试行为？

完全可以用tenacity实现这种针对不同错误类型的定制化重试逻辑，它提供的自定义重试条件、重试前钩子等功能，能完美满足你的需求。下面直接给出可运行的示例代码和关键逻辑说明：

示例代码 #

首先定义模拟的请求、令牌刷新函数，以及tenacity的核心重试逻辑：

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, RetryError
import logging

# 配置日志，便于追踪重试流程
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 模拟存储当前令牌
current_token = "expired_token"

def refresh_token():
    """模拟令牌刷新逻辑"""
    global current_token
    current_token = "new_valid_token"
    logger.info("令牌已刷新，准备重试请求")

def make_request():
    """模拟发送API请求"""
    headers = {"Authorization": f"Bearer {current_token}"}
    # 这里替换为实际请求地址，示例用模拟状态码触发错误
    response = requests.get("https://example.com/api/data", headers=headers)
    
    # 模拟不同错误场景
    if response.status_code == 401:
        raise requests.exceptions.HTTPError("401 Unauthorized", response=response)
    elif response.status_code == 500:
        raise requests.exceptions.HTTPError("500 Internal Server Error", response=response)
    response.raise_for_status()
    return response.json()

def should_retry(attempt):
    """自定义重试条件：仅在遇到401或500错误时触发重试"""
    exc = attempt.exception()
    if isinstance(exc, requests.exceptions.HTTPError):
        return exc.response.status_code in (401, 500)
    return False

def before_retry_handler(retry_state):
    """重试前的差异化处理：401刷新令牌，500记录等待日志"""
    exc = retry_state.outcome.exception()
    if isinstance(exc, requests.exceptions.HTTPError):
        status_code = exc.response.status_code
        if status_code == 401:
            refresh_token()
        elif status_code == 500:
            logger.info(f"触发500错误，等待后重试（当前重试次数：{retry_state.attempt_number}）")

# 应用tenacity装饰器配置重试规则
@retry(
    retry=should_retry,
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避等待（2s、4s、8s...最多10s）
    before_sleep=before_retry_handler
)
def call_protected_api():
    return make_request()

# 调用示例
try:
    api_result = call_protected_api()
    print("请求成功返回：", api_result)
except RetryError as e:
    print("所有重试尝试失败：", e)

关键逻辑说明 #

• 自定义重试条件：通过should_retry函数判断当前异常是否是需要重试的401/500错误，tenacity会根据这个函数的返回值决定是否重试。
• 重试前钩子：before_retry_handler会在每次重试等待前执行，针对401错误调用令牌刷新方法，针对500错误记录等待日志，实现差异化处理。
• 指数退避等待：wait_exponential针对500这类服务器临时错误，采用指数增长的等待时间，避免频繁请求加重服务器负载。
• 重试次数限制：stop_after_attempt设置最大重试次数，防止无限循环重试。

如果需要扩展更多错误类型的处理逻辑，只需要在should_retry和before_retry_handler中新增对应的状态码判断即可，tenacity的灵活性完全支持这种精细化的重试控制。

内容的提问来源于stack exchange，提问作者Eduard6421