商城首页欢迎来到中国正版软件门户

您的位置:首页 >Python Flask对接第三方API,requests库使用教程

Python Flask对接第三方API,requests库使用教程

  发布于2026-04-20 阅读(0)

扫一扫,手机访问

必须为requests.get()显式设置超时timeout=(连接超时,读取超时),捕获Timeout、ConnectionError、HTTPError等具体异常,密钥从环境变量注入,高并发下应异步化或使用连接池。

Python开发Flask后端如何对接第三方API_使用requests库发起请求

requests.get() 超时没设,服务卡死在第三方响应上

Flask 是同步框架,requests.get() 默认不设超时会无限等待,一旦第三方 API 响应慢或挂掉,整个 Flask worker 就卡住,请求堆积、CPU 占满、后续所有接口都不可用。

必须显式设置 timeout 参数,且建议拆成连接超时和读取超时:

  • timeout=(3, 10) 表示 3 秒连不上就放弃,连上后最多等 10 秒收完数据
  • 不要只写 timeout=10 —— 这只限制总耗时,网络不通时仍可能卡满 10 秒,不如分开控制更稳
  • 生产环境建议连接超时 ≤ 3 秒,读取超时 ≤ 15 秒;金融类接口可再收紧

示例:

try:
    resp = requests.get("https://api.example.com/data", timeout=(3, 10))
    resp.raise_for_status()
except requests.exceptions.Timeout:
    return {"error": "第三方服务响应超时"}, 504
except requests.exceptions.HTTPError as e:
    return {"error": f"HTTP {resp.status_code}"}, 502

Flask 视图里直接调 requests,没做异常隔离

第三方 API 出错(DNS 失败、SSL 验证失败、连接被拒)如果没兜住,会抛出未捕获异常,导致 Flask 返回 500,还可能暴露堆栈信息。

要覆盖常见异常类型,不能只 catch Exception

  • requests.exceptions.ConnectionError:DNS 解析失败、目标地址不可达、防火墙拦截
  • requests.exceptions.Timeout:已提过,必须单独处理
  • requests.exceptions.TooManyRedirects:对方重定向环路,别让它拖垮你的服务
  • requests.exceptions.SSLError:证书问题,测试环境可能忽略,但生产务必校验或明确降级逻辑

注意:resp.raise_for_status() 会在状态码 ≥400 时抛 HTTPError,它属于 RequestException 子类,也要包进 try 块里。

headers 和 auth 写死在代码里,上线后密钥泄露风险高

Authorization token 或 API-Key 直接写在 Python 文件里,Git 提交、日志打印、错误堆栈都可能暴露。

正确做法是运行时注入:

  • 从环境变量读:os.getenv("THIRD_PARTY_API_KEY"),启动 Flask 时用 export THIRD_PARTY_API_KEY=xxx
  • python-decoupledynaconf 管理配置,开发/生产环境自动切换
  • Header 中敏感字段如 AuthorizationX-API-Key,务必检查是否误打日志 —— 别在 app.logger.info(f"req headers: {headers}") 里裸打

示例安全写法:

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {os.getenv('THIRD_PARTY_TOKEN', '')}"
}

并发量上来后,requests 同步阻塞拖垮 Flask 吞吐

一个请求调一次第三方 API,QPS 到 50 就可能把 Flask 的线程池占满,尤其当第三方响应延迟波动大时。

短期缓解可以加连接池复用,但治标不治本:

  • requests.Session()HTTPAdapter,复用 TCP 连接:
    session = requests.Session()
    adapter = requests.adapters.HTTPAdapter(pool_connections=10, pool_maxsize=20)
    session.mount("https://", adapter)
  • 长期必须考虑异步化:用 httpx.AsyncClient + async/await 视图(需 Flask 2.0+ + enable_async=True),或把调用下沉到 Celery 异步任务
  • 别自己手写线程池 —— Flask 的线程模型和 Gunicorn 的工作进程模型容易冲突,反而更难 debug

真实压测中,同步 requests 在 30+ QPS 下平均响应时间就可能翻倍,这点常被低估。

第三方 API 的不确定性远高于本地逻辑,超时、异常类型、凭证管理、连接复用,每个点松动一点,线上就是雪崩前兆。最麻烦的不是写不出请求,而是没想清楚“它挂了你的服务还能不能活”。

本文转载于:互联网 如有侵犯,请联系zhengruancom@outlook.com删除。
免责声明:正软商城发布此文仅为传递信息,不代表正软商城认同其观点或证实其描述。

热门关注