脚本怎样灵活配置接口重试次数

wen 实用脚本 2026-06-25 31

本文目录导读：

脚本怎样灵活配置接口重试次数

环境变量配置（最灵活）
配置文件方式（适合复杂项目）
装饰器模式（最推荐）
动态重试配置（最灵活）
命令行参数方式
最佳实践建议

对于脚本中灵活配置接口重试次数,有几种常见且实用的方法，我将从简单到复杂为你介绍，并提供代码示例。

环境变量配置（最灵活）

通过环境变量控制重试次数,无需修改代码。

import os
import time
import requests
from functools import wraps
# 从环境变量获取重试次数，默认3次
MAX_RETRIES = int(os.environ.get('API_RETRY_TIMES', '3'))
RETRY_DELAY = int(os.environ.get('API_RETRY_DELAY', '1'))  # 秒
def retry_api_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        max_retries = kwargs.pop('max_retries', MAX_RETRIES)
        retry_delay = kwargs.pop('retry_delay', RETRY_DELAY)
        for attempt in range(max_retries):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                print(f"尝试 {attempt + 1}/{max_retries} 失败: {e}")
                time.sleep(retry_delay * (attempt + 1))  # 递增延迟
    return wrapper
@retry_api_call
def call_api(url):
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()
# 使用方式
if __name__ == "__main__":
    # 运行时设置：API_RETRY_TIMES=5 python script.py
    result = call_api("https://api.example.com/data")

运行方式：

# 临时设置
API_RETRY_TIMES=5 API_RETRY_DELAY=2 python script.py
# 或导出到环境
export API_RETRY_TIMES=3
python script.py

配置文件方式（适合复杂项目）

使用 YAML 或 JSON 配置文件。

import yaml
import json
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
class APIClient:
    def __init__(self, config_file='config.yaml'):
        with open(config_file, 'r') as f:
            self.config = yaml.safe_load(f)
        self.retry_config = self.config.get('retry', {})
        self.max_retries = self.retry_config.get('max_attempts', 3)
        self.backoff_factor = self.retry_config.get('backoff_factor', 1)
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10)
    )
    def call_api(self, url, **kwargs):
        response = requests.get(url, **kwargs)
        response.raise_for_status()
        return response.json()
# config.yaml 示例
"""
retry:
  max_attempts: 5
  backoff_factor: 2
  retryable_status_codes: [500, 502, 503, 504]
  retryable_exceptions:
    - requests.exceptions.ConnectionError
    - requests.exceptions.Timeout
"""

装饰器模式（最推荐）

使用 tenacity 库实现灵活的重试策略。

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from typing import Callable, Optional
def create_retry_decorator(
    max_retries: int = 3,
    min_wait: int = 1,
    max_wait: int = 10,
    retry_on_status: list = None
) -> Callable:
    """创建可配置的重试装饰器"""
    if retry_on_status is None:
        retry_on_status = [500, 502, 503, 504]
    def retry_with_config(func):
        @retry(
            stop=stop_after_attempt(max_retries),
            wait=wait_exponential(multiplier=1, min=min_wait, max=max_wait),
            retry=retry_if_exception_type((
                requests.exceptions.ConnectionError,
                requests.exceptions.Timeout,
                requests.exceptions.HTTPError
            )),
            before_sleep=lambda retry_state: print(
                f"重试 {retry_state.attempt_number}/{max_retries}, "
                f"等待 {retry_state.next_action.sleep} 秒"
            )
        )
        def wrapper(*args, **kwargs):
            response = func(*args, **kwargs)
            if response.status_code in retry_on_status:
                raise requests.exceptions.HTTPError(
                    f"状态码 {response.status_code} 需要重试"
                )
            return response
        return wrapper
    return retry_with_config
# 使用示例
@create_retry_decorator(max_retries=5, min_wait=1, max_wait=30)
def fetch_data(url):
    return requests.get(url, timeout=10)

动态重试配置（最灵活）

运行时动态调整重试参数。

import requests
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class RetryConfig:
    max_retries: int = 3
    base_delay: float = 1.0
    max_delay: float = 60.0
    backoff_factor: float = 2.0
    jitter: bool = True
class AdaptiveRetryClient:
    def __init__(self, config: Optional[RetryConfig] = None):
        self.config = config or RetryConfig()
        self.retry_stats = {}  # 记录重试统计数据
    def call_with_retry(self, url, method='GET', **kwargs):
        """执行带重试的API调用"""
        retry_count = 0
        while retry_count <= self.config.max_retries:
            try:
                response = requests.request(method, url, **kwargs)
                if response.status_code < 500:
                    return response
                # 针对特定状态码动态调整
                if response.status_code in [429]:  # 限流
                    self.config.max_retries = min(
                        self.config.max_retries + 1, 10
                    )
            except requests.exceptions.ConnectionError:
                # 连接错误时增加基础延迟
                self.config.base_delay *= 1.5
            except requests.exceptions.Timeout:
                # 超时错误时减少重试间隔
                pass
            retry_count += 1
            delay = self._calculate_delay(retry_count)
            print(f"重试 {retry_count}/{self.config.max_retries}, 等待 {delay:.1f}秒")
            time.sleep(delay)
        raise Exception(f"API调用失败，已重试 {self.config.max_retries} 次")
    def _calculate_delay(self, attempt):
        """计算延迟时间（带指数退避和随机抖动）"""
        import random
        delay = min(
            self.config.base_delay * (self.config.backoff_factor ** (attempt - 1)),
            self.config.max_delay
        )
        if self.config.jitter:
            delay = delay * (0.5 + random.random())
        return delay
# 使用示例
client = AdaptiveRetryClient(RetryConfig(max_retries=5))
response = client.call_with_retry("https://api.example.com/data")

命令行参数方式

通过命令行参数传递重试配置。

import argparse
import requests
import sys
def main():
    parser = argparse.ArgumentParser(description='API调用脚本')
    parser.add_argument('--retries', type=int, default=3,
                       help='最大重试次数（默认：3）')
    parser.add_argument('--retry-delay', type=float, default=1.0,
                       help='重试间隔（秒，默认：1）')
    parser.add_argument('--retry-backoff', type=float, default=2.0,
                       help='退避因子（默认：2）')
    parser.add_argument('url', help='API URL')
    args = parser.parse_args()
    # 调用带重试的API
    for attempt in range(args.retries):
        try:
            response = requests.get(args.url, timeout=5)
            response.raise_for_status()
            print(f"成功: {response.json()}")
            break
        except requests.exceptions.RequestException as e:
            if attempt == args.retries - 1:
                print(f"最终失败: {e}")
                sys.exit(1)
            delay = args.retry_delay * (args.retry_backoff ** attempt)
            print(f"尝试 {attempt + 1}/{args.retries} 失败, "
                  f"{delay:.1f}秒后重试...")
            time.sleep(delay)
if __name__ == "__main__":
    main()

运行方式：

python script.py --retries 5 --retry-delay 2 https://api.example.com/data

最佳实践建议

使用环境变量：适合不同环境（开发/测试/生产）使用不同配置
指数退避：避免立即重试给服务器造成压力
添加抖动：防止多客户端同时重试
日志记录：记录重试次数和原因
针对不同错误类型：连接错误、超时、HTTP错误使用不同策略
断路器模式：连续失败时可暂停重试一段时间

选择哪种方式取决于你的具体场景：

快速脚本：使用环境变量
复杂项目：使用配置文件 + tenacity
需要统计分析：使用动态配置方式