用 requests 写一个健壮的 HTTP 小工具:重试、超时与错误处理一次讲清

举报
温华9527 发表于 2026/10/30 06:13:31 2026/10/30
【摘要】 很多 HTTP 小工具一开始都写得很简单:一行 requests.get 就去拿数据,本地跑得好好的,一到弱网、接口抖动或者对方限流就原地失败。问题往往不是接口本身,而是小工具没有给网络留出容错空间。本文用一个可复用的小函数,把重试、超时和错误处理一次讲清楚。 先想清楚三件事第一,超时必须显式设置。requests 默认没有超时,网络一旦挂起,程序就可能一直等下去。建议用元组形式分别控制连接...

很多 HTTP 小工具一开始都写得很简单:一行 requests.get 就去拿数据,本地跑得好好的,一到弱网、接口抖动或者对方限流就原地失败。问题往往不是接口本身,而是小工具没有给网络留出容错空间。本文用一个可复用的小函数,把重试、超时和错误处理一次讲清楚。

先想清楚三件事

第一,超时必须显式设置。requests 默认没有超时,网络一旦挂起,程序就可能一直等下去。建议用元组形式分别控制连接超时和读取超时,例如 timeout=(3, 10),表示连接最多等 3 秒、读取最多等 10 秒。

第二,不是所有错误都值得重试。连接中断、超时、429 和 5xx 通常是临时问题,重试有意义;400、401、403、404 这类客户端错误,重试多少次结果都一样,只会浪费时间和配额。

第三,重试要有节奏。立即连续重试等于给已经不稳定的服务继续加压,应该用指数退避加随机抖动;如果响应里有 Retry-After,能遵守就尽量遵守。

一个可复用的 fetch_json

import logging
import random
import time

import requests
from requests.adapters import HTTPAdapter

logger = logging.getLogger(__name__)

RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}


def fetch_json(url, *, params=None, max_retries=3, timeout=(3, 10)):
    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=0))
    session.mount("http://", HTTPAdapter(max_retries=0))
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            response = session.get(url, params=params, timeout=timeout)
            if response.status_code in RETRYABLE_STATUS and attempt < max_retries:
                wait = _backoff(attempt, response)
                logger.warning("第 %s 次请求返回 %s,%s 秒后重试", attempt + 1, response.status_code, wait)
                time.sleep(wait)
                continue
            response.raise_for_status()
            return response.json()
        except (requests.ConnectionError, requests.Timeout) as exc:
            last_error = exc
            if attempt < max_retries:
                wait = _backoff(attempt)
                logger.warning("第 %s 次请求失败:%s,%s 秒后重试", attempt + 1, exc, wait)
                time.sleep(wait)
                continue
            raise
    raise last_error


def _backoff(attempt, response=None):
    if response is not None:
        retry_after = response.headers.get("Retry-After")
        if retry_after and retry_after.isdigit():
            return int(retry_after)
    base = 2 ** attempt
    return base + random.uniform(0, 0.5)

这段代码有几个刻意的设计:先关掉 requests 内置的重试,改由自己控制,避免两套重试逻辑叠加;Session 复用连接,比每次新建请求更省开销;只对临时错误重试,客户端错误直接交给 raise_for_status 抛出,调用方可以按状态码分别处理。

调用时这样写

if __name__ == "__main__":
    try:
        data = fetch_json("https://example.com/api/items", params={"page": 1})
        print(data)
    except requests.HTTPError as exc:
        print("接口返回错误状态码:", exc)
    except requests.RequestException as exc:
        print("网络请求最终失败:", exc)

把 HTTPError 和其他 RequestException 分开捕获很有用:前者说明已经拿到响应但业务上不可用,后者说明连可靠响应都没有拿到,排查方向完全不同。

容易踩的坑

一是只写了重试却忘了超时,结果程序在第一次请求就卡住,重试逻辑根本没有机会执行。二是对 POST、PUT 这类可能产生副作用的请求无脑重试,万一第一次请求其实已经成功,只是响应丢了,重试就可能重复下单或重复写入,需要先确认接口是否幂等。三是把重试次数和退避时间写死在调用处,每个接口的容忍度不同,应该做成参数,方便按接口调整。

小结

一个健壮的 HTTP 小工具并不需要很复杂:超时兜底、只重试临时错误、退避加抖动、错误分层处理,再加一个可复用的 Session,就能覆盖日常大部分场景。先把这套骨架写进自己的工具箱,下次再接新接口时,直接复用它,而不是每次都从一行 requests.get 重新开始。

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0)

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。