CodeWiki

Requests

用 Python Requests 构造 HTTP 请求、处理响应、复用会话，并为超时、重试、流式读取和 TLS 验证设定清晰边界。

难度 进阶 时长 标准深度约 12分钟
版本 Python 3.14
what

Requests 是同步 HTTP 客户端。它把方法、URL、查询参数、请求头和请求体组装成请求，再把状态码、响应头与响应体封装为 Response。

when

它适合 Python 中直接、同步的 API 调用和下载任务。需要原生异步、HTTP/2 或大量并发连接时，应改用为该执行模型设计的客户端。

how

复用 Session，给每次调用设置超时，先判断 HTTP 状态再解析响应。只有确认操作可安全重放时才配置重试，并保持 TLS 验证开启。

是什么，为什么存在

Requests 是 Python 的第三方同步 HTTP 客户端。它把 URL 编码、请求头、Cookie、认证、重定向、TLS 证书验证和响应解码放进一套一致的 API，让应用代码关注 HTTP 交换本身，而不是直接操作套接字。

一次调用会产生两个值得区分的对象。PreparedRequest 是已经编码、可发送的请求；Response 则保存服务端返回的状态码、响应头、内容，以及实际发送的请求。遇到签名错误、参数编码错误或重定向问题时，这个区分很有用。

Requests 适合命令行工具、后台任务、服务间调用以及测试代码中的同步请求。它的 I/O 会阻塞当前线程，并不因为使用 Session 就变成异步。异步应用若要同时等待大量请求，通常应选择对应执行模型的客户端，而不是把同步调用散落到事件循环中。

这个库只负责客户端一侧的 HTTP 机制。它不知道某个 404 应该返回 None 还是业务错误，也不知道 POST 是否允许重试。状态解释、重试安全性、响应大小和目标 URL 的信任边界仍是应用契约的一部分。

工作原理

便捷函数如 requests.get() 最终会创建请求、准备请求并通过传输适配器发送。准备阶段会编码 params、data、json 和 files，补充可推导的请求头，并生成 PreparedRequest。发送阶段处理连接、TLS、代理、重定向和响应读取。

从输入到结果，可以按下面的顺序理解：

  1. 选择 HTTP 方法和目标 URL。
  2. 把查询参数、请求头、Cookie、认证和请求体交给 Requests。
  3. Session 合并会话级配置并生成 PreparedRequest。
  4. 已挂载的 HTTPAdapter 从连接池取连接并发送字节。
  5. Requests 收到响应头，并根据 stream 决定是否立即读取完整响应体。
  6. 调用方判断状态码、解析内容，并关闭响应或消费完响应体。

不同参数代表不同的 HTTP 位置，不能随意互换：

Requests 参数编码位置常见媒体类型或形式
params=URL 查询字符串?page=2&tag=python
headers=请求头字段Accept、Authorization
data= 字典请求体application/x-www-form-urlencoded
json=请求体application/json
files=请求体multipart/form-data

Session 保存默认请求头、Cookie、认证和适配器配置，并通过 urllib3 复用同一主机的底层连接。这个 连接池（connection pool） 只有在响应体被读完或响应明确关闭后才能顺利回收连接。会话因此同时具有配置状态和网络资源生命周期。

Response.content 返回原始字节，Response.text 使用选定编码生成字符串，Response.json() 则把响应体当作 JSON 解码。这三个接口没有一个能证明业务调用成功。应先按 API 契约检查状态码和媒体类型，再选择解析方式。

异常也分层出现。连接失败和超时属于传输问题；raise_for_status() 把 4xx、5xx 响应转成 HTTPError；JSON 解码失败会抛出 requests.exceptions.JSONDecodeError。不要用一个宽泛的 except Exception 抹掉这些差异。

示例

下面四个例子不依赖公共测试服务。第一个只准备请求，后三个启动仅监听回环地址的临时 HTTP 服务，因此输出不受外部 API 数据和网络状态影响。

检查实际发送的请求

把参数交给 Requests 比手工拼接 URL 和 JSON 更可靠。准备请求后，可以在不发送网络流量的情况下检查编码结果。

prepare_request.py
from requests import Request, Session

request = Request(
    "POST",
    "https://api.example.test/orders",
    params=[("include", "items"), ("include", "totals")],
    headers={"Accept": "application/json"},
    json={"sku": "BK-42", "quantity": 2},
)

with Session() as session:
    prepared = session.prepare_request(request)

print(prepared.method)
print(prepared.url)
print(prepared.headers["Content-Type"])
print(prepared.body.decode())
POST
https://api.example.test/orders?include=items&include=totals
application/json
{"sku": "BK-42", "quantity": 2}

重复的查询参数使用键值对列表，因而两个 include 都会保留。json= 同时执行 JSON 序列化并设置 Content-Type；如果服务端要求另一种 JSON 表示，再显式控制序列化过程。

PreparedRequest 中可能含有访问令牌、Cookie 或个人数据。诊断时只记录允许公开的字段，并对请求头和请求体做脱敏，不要把整个对象直接写进日志。

读取成功的 JSON 响应

这个本地服务回显搜索词。客户端给连接和读取分别设置一秒超时，检查状态后才调用 json()。

read_json.py
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from urllib.parse import parse_qs, urlsplit

import requests

class CatalogHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        query = parse_qs(urlsplit(self.path).query)
        body = json.dumps({"query": query["q"][0], "count": 2}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

server = ThreadingHTTPServer(("127.0.0.1", 0), CatalogHandler)
worker = Thread(target=server.serve_forever, daemon=True)
worker.start()

try:
    url = f"http://127.0.0.1:{server.server_port}/search"
    response = requests.get(
        url,
        params={"q": "blue mug", "limit": 2},
        headers={"Accept": "application/json"},
        timeout=(1, 1),
    )
    response.raise_for_status()
    print(response.request.path_url)
    print(response.headers["Content-Type"])
    print(response.json())
finally:
    server.shutdown()
    server.server_close()
/search?q=blue+mug&limit=2
application/json; charset=utf-8
{'query': 'blue mug', 'count': 2}

response.request.path_url 显示空格已编码为 +，原始字典无需自行转义。服务端同时声明了 JSON 和字符集；实际客户端还应验证媒体类型是否符合契约，而不是看到可解析的大括号就接受。

如果响应体可能很大，不要为了记录日志而调用 response.text 或 response.content，因为它们会把完整内容载入内存。使用 stream=True 时，应在 with 块内通过 iter_content() 逐块消费。

分开处理 HTTP 错误和超时

业务函数把找到的订单与 404 结果转换为两种明确返回值。传输超时走另一条路径，调用方不会把「服务端确认不存在」和「没有得到结果」混为一谈。

handle_errors.py
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
import requests

class OrderHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        status = 200 if self.path == "/orders/42" else 404
        payload = {"id": 42, "state": "paid"} if status == 200 else {"error": "order not found"}
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

def fetch_order(session, base_url, order_id):
    try:
        response = session.get(f"{base_url}/orders/{order_id}", timeout=(1, 1))
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as error:
        detail = error.response.json().get("error", "unknown error")
        return {"status": error.response.status_code, "error": detail}
    except requests.Timeout:
        return {"error": "request timed out"}

server = ThreadingHTTPServer(("127.0.0.1", 0), OrderHandler)
Thread(target=server.serve_forever, daemon=True).start()
try:
    base_url = f"http://127.0.0.1:{server.server_port}"
    with requests.Session() as session:
        print(fetch_order(session, base_url, 42))
        print(fetch_order(session, base_url, 7))
finally:
    server.shutdown()
    server.server_close()
{'id': 42, 'state': 'paid'}
{'status': 404, 'error': 'order not found'}

例子假定错误响应也是 JSON，这是被本地服务明确保证的契约。真实客户端若不能依赖这一点，应在错误分支中单独处理媒体类型和 JSONDecodeError，同时限制保留到日志中的响应片段。

这里捕获 Timeout 是因为调用方把连接超时和读取超时映射成同一种结果。需要不同恢复动作时，可以分别捕获 ConnectTimeout 与 ReadTimeout；前者通常发生在请求尚未送达时，后者可能留下未知的服务端结果。

只重试允许重放的调用

最后一个本地服务前两次返回 503，第三次返回成功。适配器只允许重试 GET，并把最大重试次数设为两次，所以总尝试次数为三次。

retry_get.py
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread

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

class BusyHandler(BaseHTTPRequestHandler):
    attempts = 0

    def do_GET(self):
        BusyHandler.attempts += 1
        status = 503 if BusyHandler.attempts < 3 else 200
        body = b"busy" if status == 503 else b"ready"
        self.send_response(status)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

retry = Retry(
    total=2,
    status_forcelist={503},
    allowed_methods={"GET"},
    backoff_factor=0,
)
server = ThreadingHTTPServer(("127.0.0.1", 0), BusyHandler)
Thread(target=server.serve_forever, daemon=True).start()
try:
    with requests.Session() as session:
        session.mount("http://", HTTPAdapter(max_retries=retry))
        url = f"http://127.0.0.1:{server.server_port}/health"
        response = session.get(url, timeout=(1, 1))
        print(response.status_code, response.text)
        print("attempts:", BusyHandler.attempts)
finally:
    server.shutdown()
    server.server_close()
200 ready
attempts: 3

生产配置通常需要非零退避和抖动，并根据 API 契约处理 Retry-After。示例把退避设为零只是为了让本地输出立即完成，不能直接复制成生产策略。

Retry 的 total 表示首次尝试之后允许的重试次数上限，而不是总调用次数。即使 HTTP 方法按规范具有 幂等性（idempotency） ，客户端仍要确认服务端实现、请求体和前置条件确实允许重放。

陷阱

省略超时或把它当成总截止时间

修复： 给每次外部调用显式传入 (connect_timeout, read_timeout)，并在更高层设置整个操作的截止时间。把重试、退避、DNS、连接和读取都计入同一个调用预算。

先解析 JSON 再判断状态

修复： 先按契约判断状态码，再检查媒体类型并解析相应模式。错误响应使用独立模式；解析失败时保留有界、脱敏的诊断信息。

无条件重试写请求

修复： 默认只重试明确允许重放的方法。必须重试创建操作时，建立真正的幂等协议，例如由调用方稳定生成键，服务端原子去重并返回同一结果。

关闭 TLS 验证来消除报错

修复： 保持默认验证。私有证书颁发机构应通过受控 CA 包配置；证书名称、有效期或链错误应修复，而不是在调用点绕过。

泄漏会话状态或流式响应

修复： 按清晰的认证和所有权边界创建会话，并用上下文管理器关闭。流式读取也放进 with，设定最大大小，正常消费或显式关闭每个响应。

请求不可信 URL

修复： 优先使用固定源站与受控路径。确需接收目标地址时，限制协议和端口，解析并验证主机与 IP，对每次重定向重新验证，并让出站网络策略承担第二层约束。

深入 超时预算不只是一个数字

超时预算不只是一个数字

连接超时覆盖建立网络连接所等待的时间，读取超时覆盖已经连接后等待下一批响应字节的时间。timeout=(2, 5) 不表示整次调用最多七秒：地址解析、多个地址尝试、重定向、持续有少量数据到达的长响应，以及重试都可能拉长总耗时。

调用链还会放大预算。上游请求若有三次尝试，而下游每次又有三次尝试，最坏路径可能触发多轮工作并占用连接。应从用户可见截止时间向内分配预算，让每层知道还剩多少时间，而不是各自采用固定的宽松超时。

Requests 没有一个参数能替代完整的墙钟截止时间。需要硬性总期限时，应在任务或进程层控制取消，并确认取消路径会关闭响应和会话。在线程中简单超时等待一个后台调用，并不等于已经停止那个网络调用。

读取超时后，服务端是否完成操作可能未知。对于读取操作，可以按策略重新获取；对于写操作，恢复流程必须依赖幂等键、状态查询或领域级对账，不能因为客户端没有响应就假定服务端没有执行。

连接复用依赖响应生命周期

Session 为同一主机复用连接，减少重复建立 TCP 与 TLS 连接的工作。它不是无限并发承诺，也不是缓存响应。池的大小、阻塞行为和适配器挂载范围都需要按服务边界配置。

默认 stream=False 会在返回前读取响应体，因此连接通常可以回到池中。stream=True 先让调用方处理响应头，响应体仍绑定着底层连接；只有读完内容或调用 close()，连接才可以可靠复用。

上下文管理器让异常路径也能关闭资源。对于下载任务，还要同时限制声明大小和实际读取大小，因为 Content-Length 可能缺失或不可信。压缩响应解码后的大小也可能远大于线上字节数。

会话保存 Cookie 和默认认证，因此复用边界不能只看主机名。两个租户访问同一 API，并不代表它们应该共享同一个会话对象。把凭据作为明确依赖传入，比在进程级单例上不断修改 headers 更容易审查。

重试必须服从 HTTP 语义

重试发生在客户端不确定前一次结果时。连接尚未建立、读取中断和服务端返回 503 看似都是「失败」，但服务端是否收到并处理请求并不相同。恢复动作应基于这个不确定性，而不是只基于异常类名称。

urllib3 的 Retry 通过 allowed_methods 和 status_forcelist 决定哪些响应重试。默认方法集合倾向于 HTTP 语义上的幂等方法，但应用仍需保证实际端点遵守该语义。一个名为 GET 却产生不可逆副作用的接口不会被客户端配置自动修正。

退避用于避免所有客户端立即重复施压，抖动用于避免它们再次同步。服务端给出 Retry-After 时，客户端应按双方契约处理，同时保留自己的最大等待和总截止时间。服务端要求等待多久都不能自动覆盖调用方的业务期限。

可重试 POST 需要端到端设计。调用方在逻辑操作开始时生成稳定幂等键，并在每次网络尝试中保持不变；服务端在提交副作用的同一一致性边界内记录键和结果。只在客户端生成随机键但每次重试都换一个值，不能去重。

输入 URL 也是安全边界

服务端程序拥有普通用户浏览器没有的网络可达性和凭据。只要不可信输入能够改变协议、主机、端口或重定向路径，就应把出站请求视为 SSRF 边界。仅检查字符串前缀很容易被用户信息段、编码差异或相似域名绕过。

安全校验需要先规范化并解析 URL，再按允许的协议、主机和端口决策。若策略禁止私有与本地地址，还要检查 DNS 解析结果，并考虑解析和连接之间地址变化的问题。最稳妥的设计仍是让调用方选择业务资源标识，由服务端映射到固定源站。

重定向会产生新的目标，因此首次 URL 通过检查不代表整个请求链安全。可以禁用自动重定向并逐跳验证，或者使用只允许访问批准源站的传输层。还应限制最大重定向次数，避免循环和不必要的资源消耗。

代理环境变量也是路由输入。部署环境若不应继承 HTTP_PROXY、HTTPS_PROXY 或 NO_PROXY，应明确配置 Session.trust_env，并在目标环境验证效果。代理既能改变网络路径，也可能看到未受端到端 TLS 保护的流量。

延伸阅读

检查点

4个问题 · 1 道输出预测题 · 1 道找错题

前置内容 后端开发入门RESTful API 设计
下一篇 HTTPX 客户端 Oauth2 guide 即将上线 Rate limiting 即将上线 Testing 即将上线
复制为 Markdown 面试题库 在 GitHub 上编辑 报告错误 讲清楚了吗？