Python 网络请求工程实践：从超时重试到证书信任的系统化解决

在写定时自动化脚本（每日签到、数据同步、健康检查、批量拉取报表）时，最容易出问题的往往不是业务逻辑，而是网络层。典型症状非常一致：脚本昨天还好好的，今天突然抛异常；或者在本机跑得好好的，换一台机器就报证书错误。这类问题看起来琐碎，却直接决定了脚本能不能长期稳定运行。

下面结合真实踩坑经历，把超时、重试、证书、幂等这四件事一次性讲清楚，并给出一个可以直接复用的模板。

一、超时：永远不要写裸的 requests.get

requests 的默认行为是永不超时。这意味着一旦服务端不返回，你的线程会一直挂住。对于定时任务，这会导致任务堆积、后续执行全部错过。

正确做法是显式给出"连接超时"与"读取超时"两个值——它们的含义完全不同：连接超时是 TCP 握手的等待时间，读取超时是两次数据包之间的最大间隔。

import requests

# 连接 3.05s，读取 10s
resp = requests.get("https://example.com/api", timeout=(3.05, 10))
resp.raise_for_status()

经验值：连接超时 3~5 秒足够（握手要么很快，要么网络已经不通），读取超时按接口的实际耗时上浮 2~3 倍来定。

二、重试：指数退避，且只对可重试的错误重试

重试不是简单地"失败就再来一次"。两个原则：一是退避，连续重试会放大对端的压力；二是区分错误类型，4xx 里的 400/401/403 重试一万次也没用，而 429/5xx 则值得重试。

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(
    total=3,                      # 总重试次数
    connect=2, read=2, status=2,
    backoff_factor=0.5,           # 间隔: 0.5s -> 1s -> 2s
    status_forcelist=(429, 500, 502, 503, 504),
    allowed_methods=frozenset(["GET", "HEAD", "OPTIONS"]),
    raise_on_status=False,
)

session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=retry))
session.mount("http://",  HTTPAdapter(max_retries=retry))

注意 allowed_methods 默认只对幂等方法生效。如果你的 POST 本身做了幂等设计（见下文），再把它加进去才安全。

三、证书：certifi 与系统 CA 的信任链不一致

这个坑非常隐蔽：同样是 requests.get(url)，在 A 机器上正常，在 B 机器上抛 SSL: CERTIFICATE_VERIFY_FAILED。原因是 requests 默认使用 certifi 自带的证书包，而某些站点（尤其是用了企业内部 CA 或较新根证书的站点）的信任链并不在 certifi 里，却在系统的 /etc/ssl/certs/ca-certificates.crt 里。

先用 openssl 看清楚对端到底返回了什么链：

openssl s_client -showcerts -servername example.com \
    -connect example.com:443 < /dev/null 2>/dev/null | grep -E "subject|issuer"

修复方式按优先级排序：

# 方案 1（推荐，作用于单次进程）：指定合并后的 CA 文件
export REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt

# 方案 2（推荐，作用于单个 Session）
session.verify = "/path/to/merged-ca.pem"

# 方案 3：把系统 CA 追加进 certifi（会改动 site-packages，谨慎）
python - <<'EOF'
import certifi
with open(certifi.where(), "a", encoding="utf-8") as f:
    f.write("\n")
    f.write(open("/etc/ssl/certs/ca-certificates.crt", encoding="utf-8").read())
EOF

不要用 verify=False 图省事——它只是把问题藏起来，还会在日志里刷一堆 InsecureRequestWarning。

四、幂等：重复运行不应该产生副作用

定时任务一定会被重复触发：手动补跑、调度器重试、上一次还没退出下一次就启动了。如果脚本不具备幂等性，重复运行就可能重复写入、重复扣款、重复发通知。

三个低成本做法：

• 给每次运行生成 run_id，写库时以它为唯一键，重复插入直接被约束挡掉；
• 写文件采用"临时文件 + 原子 rename"，避免半截文件被下游读到；
• 业务请求带上客户端生成的去重键，服务端据此返回第一次的结果。

五、一个可直接复用的模板

把上面几点组合起来，就是下面这个不到 60 行的骨架，可以直接替换掉你脚本里散落的 requests 调用。

import logging, requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

log = logging.getLogger(__name__)
TIMEOUT = (3.05, 15)
CA_BUNDLE = "/etc/ssl/certs/ca-certificates.crt"   # 或 certifi.where()

def build_session() -> requests.Session:
    s = requests.Session()
    s.headers.update({"User-Agent": "health-check/1.0"})
    retry = Retry(total=3, backoff_factor=0.5,
                  status_forcelist=(429, 500, 502, 503, 504),
                  allowed_methods=frozenset(["GET", "HEAD"]),
                  raise_on_status=False)
    s.mount("https://", HTTPAdapter(max_retries=retry))
    s.mount("http://",  HTTPAdapter(max_retries=retry))
    return s

def fetch_json(session, url, **kw):
    try:
        r = session.get(url, timeout=TIMEOUT, verify=CA_BUNDLE, **kw)
        r.raise_for_status()
        return r.json()
    except requests.exceptions.Timeout:
        log.warning("timeout: %s", url)
    except requests.exceptions.HTTPError as e:
        log.warning("http %s: %s", e.response.status_code, url)
    except requests.exceptions.RequestException as e:
        log.warning("failed %s: %s", url, e)
    return None

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    with build_session() as s:
        data = fetch_json(s, "https://example.com/api/status")
        print("ok" if data else "degraded")

这个模板的关键不在于代码多，而在于把"失败"当成常态：超时有上限、重试有退避、证书可配置、失败可降级。做到这四点，脚本才敢真正放进 crontab 里长期跑。

小结

• 显式设置 timeout=(connect, read)，不要让请求无限等待；
• 用 Retry 做指数退避，并只对可恢复的错误重试；
• 证书错误先排查信任链，用 REQUESTS_CA_BUNDLE 或 session.verify 修复，别用 verify=False；
• 把幂等作为设计前提，而不是事后补救。

如果你也在维护类似的定时任务，欢迎在评论区交流踩过的坑。