Python API调用避坑指南:超时、重试、日志完整实战

功能模块

选型方案

优势说明

HTTP请求

requests

Python最通用HTTP客户端，语法简洁、生态成熟

重试机制（生产首选）

urllib3.Retry + HTTPAdapter

原生支持、无需额外安装依赖、稳定无bug

重试机制（灵活首选）

tenacity

语法简洁、配置灵活、支持精细化重试规则

日志记录

logging标准库

无需额外依赖、日志分级、可对接日志系统

超时控制

requests timeout参数

精准控制连接、读取超时，避免阻塞

python                  import requests                  from requests.adapters import HTTPAdapter                  from urllib3.util.retry import Retry                  import logging                  # 初始化日志配置                  logging.basicConfig(                  level=logging.INFO,                  format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"                  )                  logger = logging.getLogger("APIClient")                  class APIClient:                  def __init__(self, base_url, timeout=30, max_retries=3, backoff_factor=1.0):                  self.base_url = base_url.rstrip('/')                  self.timeout = timeout                  # 创建持久化Session，复用连接，提升请求效率                  self.session = requests.Session()                  # 核心重试策略配置                  retry_strategy = Retry(                  total=max_retries,# 最大重试总次数                  backoff_factor=backoff_factor,# 指数退避因子                  status_forcelist=[429, 500, 502, 503, 504],# 触发重试的错误状态码                  allowed_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD"],# 支持重试的请求方法                  raise_on_redirect=False,                  raise_on_status=False                  )                  # 挂载重试适配器到HTTP/HTTPS请求                  adapter = HTTPAdapter(max_retries=retry_strategy)                  self.session.mount("http://", adapter)                  self.session.mount("https://", adapter)                  def request(self, method, endpoint, **kwargs):                  # 拼接完整请求地址                  url = f"{self.base_url}/{endpoint.lstrip('/')}"                  # 优先级：用户传入超时 > 默认超时                  kwargs.setdefault("timeout", self.timeout)                  try:                  logger.info(f"发送 {method} 请求：{url}")                  response = self.session.request(method, url, **kwargs)                  response.raise_for_status()# 非2xx状态码主动抛异常                  logger.info(f"请求成功，状态码：{response.status_code}")                  return response.json()                  except requests.exceptions.Timeout as e:                  logger.error(f"请求超时异常：{str(e)}")                  raise                  except requests.exceptions.ConnectionError as e:                  logger.error(f"网络连接异常：{str(e)}")                  raise                  except requests.exceptions.HTTPError as e:                  logger.error(f"HTTP请求错误：{str(e)}，响应内容：{e.response.text if e.response else '无'}")                  raise                  except Exception as e:                  logger.error(f"未知请求异常：{str(e)}")                  raise                  # 封装常用GET请求                  def get(self, endpoint, params=None, **kwargs):                  return self.request("GET", endpoint, params=params, **kwargs)                  # 封装常用POST请求                  def post(self, endpoint, data=None, json=None, **kwargs):                  return self.request("POST", endpoint, data=data, json=json, **kwargs)                  # 调用示例                  if __name__ == "__main__":                  # 初始化客户端，自定义超时、重试次数、退避因子                  client = APIClient("https://api.example.com", timeout=5, max_retries=3, backoff_factor=0.5)                  try:                  result = client.get("/users/1")                  print("接口返回数据：", result)                  except Exception:                  logger.exception("API接口调用最终失败")                  

bash                  pip install tenacity

python                  import requests                  from tenacity import (                  retry,                   stop_after_attempt,                   wait_exponential,                   retry_if_exception_type,                   before_log,                   after_log                  )                  import logging                  # 日志初始化                  logging.basicConfig(level=logging.INFO)                  logger = logging.getLogger(__name__)                  # 定义需要重试的异常类型（仅临时可恢复错误）                  RETRY_EXCEPTIONS = (                  requests.exceptions.Timeout,# 超时重试                  requests.exceptions.ConnectionError # 网络波动重试                  )                  # 装饰器配置重试规则                  @retry(                  stop=stop_after_attempt(3),# 最大重试3次                  wait=wait_exponential(multiplier=1, min=1, max=10),# 指数退避等待1-10秒                  retry=retry_if_exception_type(RETRY_EXCEPTIONS),# 仅匹配指定异常才重试                  before=before_log(logger, logging.INFO),# 重试前打印日志                  after=after_log(logger, logging.INFO),# 重试后打印日志                  reraise=True# 重试失败后抛出原始异常                  )                  def call_api(url, timeout=30, **kwargs):                  """通用API请求函数"""                  logger.info(f"发起接口请求：{url}")                  resp = requests.get(url, timeout=timeout, **kwargs)                  # 主动拦截服务端错误状态码，触发重试                  if resp.status_code in [429, 500, 502, 503, 504]:                  raise requests.exceptions.HTTPError(f"服务端异常 状态码：{resp.status_code}", response=resp)                  resp.raise_for_status()                  return resp.json()                  # 调用示例                  if __name__ == "__main__":                  try:                  data = call_api("https://api.example.com/users/1", timeout=5)                  print("接口数据：", data)                  except Exception as e:                  logger.error(f"接口调用最终失败：{str(e)}")

python                  import requests                  import logging                  import time                  logger = logging.getLogger(__name__)                  def robust_api_request(method, url, max_retries=3, timeout=30, backoff_factor=1, **kwargs):                  """                  零依赖、高可用API请求函数                  :param method: HTTP请求方法                  :param url: 接口完整地址                  :param max_retries: 最大重试次数                  :param timeout: 请求超时时间                  :param backoff_factor: 指数退避因子                  :return: 接口JSON数据                  """                  for attempt in range(1, max_retries + 1):                  try:                  logger.info(f"第 {attempt}/{max_retries} 次请求：{method} {url}")                  resp = requests.request(method, url, timeout=timeout, **kwargs)                  # 区分可重试错误和不可重试错误                  if resp.status_code >= 400:                  # 5xx服务端错误、429限流：可重试                  if resp.status_code in (429, 500, 502, 503, 504):                  raise requests.exceptions.HTTPError(f"服务端临时故障 状态码：{resp.status_code}", response=resp)                  # 4xx客户端错误：直接失败，不重试                  else:                  resp.raise_for_status()                  logger.info(f"请求成功，耗时：{resp.elapsed.total_seconds():.2f}秒")                  return resp.json()                  # 网络超时、连接失败：重试                  except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:                  logger.warning(f"临时请求失败：{str(e)}")                  if attempt == max_retries:                  logger.error("已达最大重试次数，请求彻底失败")                  raise                  # 计算退避等待时间                  sleep_time = backoff_factor * (2 ** (attempt - 1))                  logger.info(f"等待 {sleep_time:.2f} 秒后重试...")                  time.sleep(sleep_time)                  # HTTP状态码错误处理                  except requests.exceptions.HTTPError as e:                  # 客户端错误直接抛出，不重试                  if e.response and 400 <= e.response.status_code < 500 and e.response.status_code != 429:                  logger.error(f"客户端参数错误 {e.response.status_code}，不重试")                  raise                  # 服务端错误重试                  logger.warning(f"服务端异常：{str(e)}")                  if attempt == max_retries:                  raise                  sleep_time = backoff_factor * (2 ** (attempt - 1))                  time.sleep(sleep_time)                  # 未知异常直接终止                  except Exception as e:                  logger.error(f"未知请求异常：{str(e)}")                  raise                  # 测试503故障接口重试                  if __name__ == "__main__":                  logging.basicConfig(level=logging.INFO)                  result = robust_api_request("GET", "https://httpbin.org/status/503", max_retries=4, backoff_factor=0.5)                  print(result)