# Requests

Source: https://codewiki.com/zh/backend/requests/

> - **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. 调用方判断状态码、解析内容，并关闭响应或消费完响应体。

```mermaid
flowchart LR
    A[Request arguments] --> B[Session]
    B --> C[PreparedRequest]
    C --> D[HTTPAdapter]
    D --> E[Server]
    E --> F[Response]
    F --> G[Status check]
    G --> H[Decode or stream body]
```

不同参数代表不同的 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 更可靠。准备请求后，可以在不发送网络流量的情况下检查编码结果。

<!-- quick -->

```python
# file: 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())
```

```text
POST
https://api.example.test/orders?include=items&include=totals
application/json
{"sku": "BK-42", "quantity": 2}
```

<!-- /quick -->

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

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

### 读取成功的 JSON 响应

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

```python
# file: 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()
```

```text
/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` 结果转换为两种明确返回值。传输超时走另一条路径，调用方不会把「服务端确认不存在」和「没有得到结果」混为一谈。

```python
# file: 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()
```

```text
{'id': 42, 'state': 'paid'}
{'status': 404, 'error': 'order not found'}
```


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

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

### 只重试允许重放的调用

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

```python
# file: 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()
```

```text
200 ready
attempts: 3
```

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

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

## 陷阱

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

> **陷阱:** Requests 默认不会超时。传入一个数字会同时设置连接和读取超时，但读取超时衡量的是套接字多久没有收到字节，并不是整个下载必须在该秒数内结束。

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

### 先解析 JSON 再判断状态

> **陷阱:** `response.json()` 可以成功解析 `500` 的错误响应，也可能在 `204`、HTML 网关错误或截断内容上失败。JSON 可解析与业务成功是两个独立事实。

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

### 无条件重试写请求

> **陷阱:** 把 `allowed_methods=None` 复制进 `Retry` 会让任何方法都可能重放。若 `POST` 已经创建订单，但响应在途中丢失，第二次调用可能再创建一个订单。

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

### 关闭 TLS 验证来消除报错

> **陷阱:** `verify=False` 会停止验证服务端证书，使客户端无法确认连接对象。隐藏 `InsecureRequestWarning` 只会藏掉信号，不会恢复身份验证。

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

### 泄漏会话状态或流式响应

> **陷阱:** 长期全局 `Session` 可能把 Cookie、认证或默认请求头带到不相干的调用。`stream=True` 的响应若既没读完也没关闭，还会占住连接，削弱池的复用能力。

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

### 请求不可信 URL

> **陷阱:** 把用户输入直接传给 `requests.get()` 可能造成服务端请求伪造（server-side request forgery，SSRF）。攻击者可能访问回环地址、私有网络、云元数据端点，或借重定向绕过首次检查。

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

<!-- deep -->

## 超时预算不只是一个数字

连接超时覆盖建立网络连接所等待的时间，读取超时覆盖已经连接后等待下一批响应字节的时间。`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 保护的流量。

<!-- /deep -->

[检查点: backend/requests](https://codewiki.com/zh/backend/requests/#checkpoint)

## 延伸阅读

- [Requests 快速入门](https://requests.readthedocs.io/en/latest/user/quickstart/)
- [Requests 高级用法](https://requests.readthedocs.io/en/latest/user/advanced/)
- [Requests 开发者接口](https://requests.readthedocs.io/en/latest/api/)
- [urllib3 `Retry` 参考](https://urllib3.readthedocs.io/en/stable/reference/urllib3.util.html#urllib3.util.Retry)