17370845950

应使用 requests.Session 统一管理连接复用、默认 headers、timeout 和重试策略，封装 URL 构建、参数序列化、错误映射及响应解析，并用 dataclass 或 Pydantic 约束数据结构，确保类型安全与可维护性。

用 requests.Session 管理连接复用和默认配置

直接每次调用都新建 requests.get() 会导致 TCP 连接无法复用，高并发下容易触发连接耗尽或响应变慢。用 Session 是最轻量又有效的统一入口基础。

它能自动复用底层连接、持久化 cookies、预设 headers 和 timeout，也方便后续统一加日志、重试、鉴权逻辑。

• 所有 API 调用应基于同一个 Session 实例，而不是每次 new 一个
• 通过 session.headers.update(...) 设置通用 header（如 Content-Type、Authorization）
• 务必显式设置 timeout=(3, 7) —— 第一个数字是 connect timeout，第二个是 read timeout；不设 timeout 可能卡死整个调用链
• 避免在 Session 上直接设 auth=...，除非所有接口共用同一套认证；更推荐在每个请求前动态注入 token

import requests
session = requests.Session()
session.headers.update({
"Content-Type": "application/json",
"User-Agent": "my-api-client/1.0"
})
session.timeout = (3, 7)  # 注意：timeout 需在 request() 时传入，不能直接赋值给 session

把 URL 拼接、参数序列化、错误解析抽成可复用函数

REST API 封装最容易散落的三处脏代码：拼 base_url + endpoint、处理 query/body 参数类型、对不同 HTTP 状态码做不同异常映射。这些必须收口。

比如 params 里传了 None 或嵌套 dict，默认 requests 不会帮你过滤或扁平化；400/422 返回的 error detail 格式不统一，直接抛原生 HTTPError 对业务层不友好。

• 写一个 build_url(base, endpoint, **kwargs)，自动处理 trailing slash、path join、URL encode
• 用 json.dumps(..., separators=(',', ':')) 控制 body 序列化格式，避免空格干扰签名或缓存
• 对 response.status_code 做集中判断：2xx 正常返回 response.json()；4xx 抛自定义 ClientError；5xx 抛 ServerError；非 JSON 响应体要 fallback 到 response.text
• 不要依赖 response.raise_for_status() —— 它不区分 4xx 和 5xx，且无法附带业务错误信息

用数据类（dataclass 或 pydantic.BaseModel）约束请求/响应结构

手写 dict 构造请求体、再用 .get('xxx', {}).get('yyy') 解析响应，是后期维护灾难的起点。类型即文档，也便于 IDE 补全和 mypy 检查。

如果项目已用 pydantic，优先用 BaseModel；否则 @dataclass + field(default_factory=dict) 更轻量。

• 请求参数封装成 RequestPayload 类，用 .model_dump(exclude_none=True)（pydantic v2）或 .dict(exclude_none=True)（v1）自动过滤空字段
• 响应统一反序列化为 ResponseData，避免裸 dict 访问引发 KeyError
• 不要在 dataclass 中写业务逻辑方法，保持纯数据载体职责；行为逻辑放在 service 层
• 注意 pydantic 的 alias 和 serialization_alias，适配前后端字段命名差异（如 user_id ↔ userId）

别让重试逻辑污染业务代码 —— 用装饰器或 session mount 统一处理

网络抖动、临时限流、DNS 缓存失效都会导致偶发失败。每处都写 for i in range(3): try ... except ... sleep(...) 既重复又难维护。

requests 本身不内置重试，但可通过 urllib3.Retry + HTTPAdapter 注入到 Session，比装饰器更底层、更干净。

• 只对幂等方法（GET、HEAD、OPTIONS）开启自动重试；POST/PUT 默认不重试，除非你明确实现了幂等 key（如 X-Idempotency-Key）
• 配置 status_forcelist=(429, 502, 503, 504)，避免对 400/401 这类客户端错误重试
• 用 backoff_factor 控制退避时间，不要硬写 time.sleep(1)
• 重试日志必须打到 debug 级别，并带上原始 URL 和 retry count，否则线上排查时看不到“到底重试了没”

from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
retry_strategy = Retry(
total=3,
status_forcelist=[429, 502, 503, 504],
backoff_factor=0.3,
allowed_methods=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)

真正难的不是封装一层函数，而是让所有开发者习惯从统一入口发请求、信任返回类型、不绕过重试和超时控制。一旦某人开始手写 requests.post(url, json={...})，那一整套设计就从根上松动了。