做Python后端、爬虫、数据对接的小伙伴,大概率都遇到过这些问题:
明明代码没Bug,调用第三方API却偶尔报错;网络突然抖动、服务端临时崩溃、请求超时,导致程序直接中断、数据丢失。
其实绝大多数偶发性API请求失败,都不是代码逻辑问题,而是网络和服务端的临时故障。想要提升程序健壮性,不用重构代码,核心只需要做好两件事:合理的重试机制 + 完整的日志记录。
今天就给大家分享两套成熟的Python API重试方案,搭配标准化日志配置,直接开箱即用,适配所有生产环境!
一、基础铺垫:带超时&异常捕获的原生请求
很多新手写的API请求,没有超时限制、没有异常细分捕获,一旦遇到问题就直接崩盘。我们先搭建一个稳固的基础请求模板,精准捕获各类请求错误。
核心功能
1.设置请求超时,避免程序卡死
2.细分超时、连接错误、HTTP状态码错误、未知异常
3.基础日志打印,快速定位问题
python import requests import logging # 基础日志全局配置 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def call_api(url, timeout=5): try: # 发起GET请求,限制超时时间 resp = requests.get(url, timeout=timeout) # 非200状态码主动抛出异常 resp.raise_for_status() return resp.json() except requests.exceptions.Timeout: logger.error("请求超时: %s", url) raise except requests.exceptions.ConnectionError: logger.error("连接错误: %s", url) raise except requests.exceptions.HTTPError as e: logger.error("HTTP错误 %s: %s", e.response.status_code, url) raise except Exception as e: logger.exception("未知错误: %s", e) raise |
这个基础版本解决了请求无限制、错误无区分的问题,但面对临时故障,只能被动报错,无法自动恢复,接下来我们加入核心的重试机制。
二、两套主流重试方案,适配不同场景
方案一:urllib3 Retry + HTTPAdapter(轻量首选)
这是Python官方生态最适配的重试方案,无需额外安装第三方库,依托requests原生适配器,专门适配REST API请求,自动处理5xx服务端故障、连接失败等问题。
核心优势
1.零额外依赖,轻量化无侵入
2.支持指数退避算法,避免批量请求压垮服务端
3.可自定义重试状态码、请求方法、重试次数
python import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry import logging logger = logging.getLogger(__name__) def create_retry_session( retries=3, backoff_factor=1,# 指数退避因子:等待时间=1*(2^(重试次数-1)) status_forcelist=(500, 502, 503, 504),# 需要重试的服务端错误码 allowed_methods={"GET", "POST"}# 允许重试的请求方法 ): # 创建持久化请求会话 session = requests.Session() # 定义重试策略 retry_strategy = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=status_forcelist, allowed_methods=allowed_methods, raise_on_status=False ) # 挂载重试适配器 adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session # 带重试机制的API请求方法 def call_with_retry(url, timeout=5): session = create_retry_session(retries=3) try: resp = session.get(url, timeout=timeout) resp.raise_for_status() logger.info("请求成功: %s", url) return resp.json() except Exception as e: logger.error("最终失败(已重试%d次): %s", session.get_adapter(url).max_retries.total, e) raise |
注意:该方案默认不重试超时异常,仅处理连接错误、5xx服务端故障,适合大部分稳定的后端API对接场景。
方案二:tenacity 通用重试(灵活全能,生产首选)
如果你的项目需要重试超时异常、自定义重试条件、精细化日志,tenacity 是目前最优解,也是企业级项目的主流选择,支持任意场景的重试规则配置。
第一步:安装依赖
bash pip install tenacity |
第二步:完整重试代码
python from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests import logging logger = logging.getLogger(__name__) # 精细化重试配置 @retry( stop=stop_after_attempt(3),# 最大重试3次 wait=wait_exponential(multiplier=1, min=1, max=10),# 指数退避等待 # 仅重试指定异常 retry=retry_if_exception_type( (requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.HTTPError) ), # 每次重试前打印警告日志 before_sleep=lambda retry_state: logger.warning( "第 %s 次重试,错误: %s", retry_state.attempt_number, retry_state.outcome.exception() ) ) def call_api_with_tenacity(url, timeout=5): resp = requests.get(url, timeout=timeout) resp.raise_for_status() return resp.json() # 调用测试 try: data = call_api_with_tenacity("https://api.example.com/data") logger.info("成功获取数据: %s", data) except Exception as e: logger.error("所有重试均失败: %s", e) |
核心优势
1.支持超时异常重试,弥补urllib3方案短板
2.可自定义重试条件(甚至支持根据返回内容判断重试)
3.丰富的等待策略,规避请求扎堆冲击服务端
三、生产级日志配置:终端+文件双输出
调试时看终端日志,线上运行需要留存日志文件排查问题。下面给大家一套可直接上线的日志配置,同时输出简洁终端日志+详细文件日志。
python import logging.config # 全局日志配置 LOGGING_CONFIG = { 'version': 1, 'formatters': { # 简洁格式:终端输出 'default': { 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s', }, # 详细格式:日志文件留存(含代码行号,方便排错) 'detailed': { 'format': '%(asctime)s - %(name)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s', } }, 'handlers': { 'console': { 'class': 'logging.StreamHandler', 'level': 'INFO', 'formatter': 'default', }, 'file': { 'class': 'logging.FileHandler', 'filename': 'api_errors.log', 'level': 'WARNING', 'formatter': 'detailed', 'encoding': 'utf-8' } }, 'root': { 'level': 'INFO', 'handlers': ['console', 'file'] } } # 加载日志配置 logging.config.dictConfig(LOGGING_CONFIG) |
日志核心规范(生产必守)
1.记录每一次重试,精准定位临时故障频率
2.记录最终失败信息,包含重试次数、原始错误
3.严禁记录敏感信息:API密钥、令牌、用户隐私数据
4.分级日志:正常请求INFO、异常重试WARNING、最终失败ERROR
四、生产环境最佳实践,避坑关键点
1. 精细化设置超时时间
区分连接超时和读取超时,避免单一超时导致的误判:
connect超时(3s):建立TCP连接的最大时间
read超时(10s):等待服务端返回数据的最大时间
示例:requests.get(url, timeout=(3, 10))
2. 优先使用指数退避+抖动
固定间隔重试容易导致所有客户端同时请求服务端,引发雪崩。指数退避策略会逐步拉长重试间隔,大幅降低服务端压力。
3. 严格区分幂等/非幂等请求
可安全重试:GET、PUT、DELETE(幂等请求,多次请求无副作用)
谨慎重试:POST请求(可能重复创建数据,需接口支持幂等才可重试)
4. 429限流异常特殊处理
遇到429请求过多报错,不要盲目重试,优先读取响应头Retry-After,等待指定时间后再重试,避免被封禁。
5. 异步场景适配
异步项目推荐搭配:httpx + tenacity 或aiohttp + backoff,同步异步场景全覆盖。
五、终极完整版代码:超时+重试+日志+自定义异常
整合所有功能,适配GET/POST请求、精准重试判断、完整日志,直接复制可用!
python import requests from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception, before_sleep_log import logging # 初始化日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def is_retryable_exception(exception): """自定义重试判断规则:仅临时故障重试""" if isinstance(exception, requests.exceptions.Timeout): return True if isinstance(exception, requests.exceptions.ConnectionError): return True if isinstance(exception, requests.exceptions.HTTPError): # 只重试5xx服务端错误,不重试4xx客户端错误 return exception.response is not None and 500 <= exception.response.status_code < 600 return False @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception(is_retryable_exception), before_sleep=before_sleep_log(logger, logging.WARNING), reraise=True ) def robust_api_call(url, payload=None, method='GET', timeout=(3, 10)): """健壮的API通用请求方法""" with requests.Session() as session: if method.upper() == 'GET': resp = session.get(url, params=payload, timeout=timeout) else: resp = session.post(url, json=payload, timeout=timeout) resp.raise_for_status() return resp.json() # 测试调用 if __name__ == "__main__": try: # 测试500故障接口,自动触发重试 data = robust_api_call("https://httpbin.org/status/500", timeout=2) print(data) except Exception as e: logger.error("所有重试结束,请求最终失败: %s", e) |
六、方案选型总结
1.轻量简单场景:优先 urllib3.Retry + HTTPAdapter,零依赖、代码简洁,满足常规REST API需求
2.生产复杂场景:首选 tenacity,支持超时重试、自定义规则、精细化日志,灵活性拉满
一套合理的重试+日志方案,能解决90%以上的API偶发故障,不用再为临时网络问题反复改代码、排查报错,大幅提升项目稳定性!