1. 项目背景
业务场景
某消息推送平台需要向 10 万个用户发送个性化短信——每个用户的手机号、姓名、套餐内容不同，但 API 端点、认证方式、HTTP 头完全一致。最初方案是循环调用requests.post(url, json=user_data, headers=auth_headers)——10 万次循环。运行后发现：每次requests.post()都在内部重新做一次完整的请求编码——URL 拼接、参数编码、headers 合并、body 序列化。其中 95% 的工作是完全重复的（同一 URL、同一认证头、同一请求体结构），只有user_data中的几个字段在变化。
更糟糕的是：在日志排查时发现，某些请求的 headers 和预期不同——可能是 requests 内部参数解析逻辑在特定边界条件下产生了非预期的行为。但由于请求是"黑盒"发出去的，无法在发送前审计最终的 HTTP 报文内容。
痛点
问题一：请求构建与发送耦合。requests.get(url, params=..., headers=...)把"构建请求"和"发送请求"两个步骤合为一体。在批量场景下，每次调用都在重复构建相同的部分，浪费 CPU 和内存。如果能将请求"预编译"成不可变的 PreparedRequest 对象，后续发送时只需替换变化的参数即可。
问题二：PreparedRequest 的可见性优势。PreparedRequest 是发送前的最终状态——所有隐式头（Host、User-Agent、Accept-Encoding 等）都已经添加、参数已经编码、body 已经序列化。通过审计 PreparedRequest，可以 100% 确定"到底发出了什么"，这对于安全审计和问题排查至关重要。
问题三：请求签名场景的黄金窗口。许多 API 需要对请求做签名（HMAC-SHA256、AWS Signature V4）。签名的输入是"发送前的最终请求"。如果在构建请求后、发送前能拦截 PreparedRequest 并插入签名逻辑，就能完美解决签名问题。而Session.send(prepared)正是这个黄金窗口。
PreparedRequest 完整链路：
requests.get(url, params={...}, headers={...}, json={...}) │ ├─ 构建 Request 对象 │ Request(method, url, params, headers, json, ...) │ ├─ 准备 PreparedRequest │ session.prepare_request(Request) │ ├─ prepare_method() # 大写化方法名 │ ├─ prepare_url(url, params) # 拼接 URL + 编码参数 │ ├─ prepare_headers(headers) # 合并 Session 默认头 + 隐式头 │ ├─ prepare_body(data, json, files) # 序列化请求体 │ └─ prepare_cookies(cookies) # 提取匹配的 Cookie │ ├─ 【这里可以插入签名/修改逻辑！】 │ └─ 发送 PreparedRequest session.send(prepared)2. 项目设计
小胖（看着终端里 10 万次循环的慢速输出）：
“大师！我们的通知发送服务太慢了——10 万条消息，每条都要调requests.post()，10 万次循环就是 10 万次请求编码。光构建请求就花了 5 秒！”
大师（查看代码，指了指requests.post(url, json=data)）：
“你每次都在让 requests 从零构建整个请求——URL 编码、headers 合并、body 序列化。但这些步骤中 95% 都是一样的！你应该用 PreparedRequest——把请求模板"预编译"一次，后续只替换变化的字段。”
小胖（挠头）：
“PreparedRequest？这是什么？我在文档里好像没见过。”
大师：
“PreparedRequest 是 requests 内部请求构建-发送分离模型的体现。你可以理解为：Request 是原料（面粉、鸡蛋、牛奶），PreparedRequest 是烤好的面包——已经成形、不可改变。你只要把面包放进烤箱（session.send()）就行。”
fromrequestsimportRequest,Session session=Session()# 1. 构建请求模板（只做一次）template=Request("POST","https://api.push.com/send",headers={"Authorization":"Bearer xxx","Content-Type":"application/json"},)# 2. 预编译为 PreparedRequest# 注意：prepared.body 此时为空（还没设数据）prepared=session.prepare_request(template)# 3. 批量发送——只改 bodyimportjsonforuserinusers:prepared.body=json.dumps(user).encode("utf-8")response=session.send(prepared,timeout=5)小白（推了推眼镜，追问）：
“prepare_request()都做了什么？为什么能节省时间？”
大师（在白板上画出流程图）：
"prepare_request()内部执行五个步骤——这就是requests.get()在内部做的事情。当我们批量发送时，需要变化的通常只是 body 或 params，其他步骤都是重复的。
而且更重要的是——PreparedRequest 是可见的。你可以在发送前查看它："
prepared=session.prepare_request(Request("GET",url,params={"q":"test"}))print(prepared.method)# GETprint(prepared.url)# https://api.example.com/search?q=testprint(dict(prepared.headers))# 所有头（包括隐式头！）print(prepared.body)# None（GET 请求没有 body）小胖：
“那不就跟 curl 的 --libcurl 一样？我可以看到最终发了什么出去？”
大师：
“正是！而且你甚至可以写一个函数把 PreparedRequest 转成 curl 命令——这在调试和与团队沟通时极其有用。”
defto_curl(prepared):"""将 PreparedRequest 转为 curl 命令"""cmd=f"curl -X{prepared.method}"fork,vinprepared.headers.items():cmd+=f' -H "{k}:{v}"'ifprepared.body:cmd+=f" -d '{prepared.body.decode()}'"cmd+=f" '{prepared.url}'"returncmd小白（眼睛一亮）：
“那签名的场景呢？我们有个 API 需要在 header 里加 HMAC 签名。”
大师：
“PreparedRequest 就是签名的完美窗口——在prepare_request()之后、send()之前，PreparedRequest 已经包含了签名的所有输入（method + url + body + timestamp）。你只需要计算签名后加到 headers 里，然后发送。”
| 生活比喻 | 技术映射 |
|---|---|
| 烤面包（一次烤很多片） | PreparedRequest 预编译 |
| 半成品披萨（撒料不同） | 相同模板 + 不同 body |
| 看快递面单（确认信息正确） | 审计 PreparedRequest |
| 在包裹上加签名封条 | prepare -> 签名 -> send |
| 转成快递单号给别人查 | PreparedRequest 转 curl 命令 |
3. 项目实战
环境准备
pipinstallrequests分步实现
步骤一：PreparedRequest 全链路解剖
目标：手动执行prepare_request的每个步骤，理解内部流程。
importrequestsfromrequestsimportRequest,Session session=Session()session.headers.update({"X-API-Key":"default-key"})# ===== Step 1: 构建 Request（原料） =====req=Request("POST","https://httpbin.org/post",params={"source":"web"},# URL 参数headers={"X-Request-ID":"req-001"},# 自定义头json={"user":"小胖","action":"test"},# JSON 请求体)# ===== Step 2: 预编译 =====prepared=session.prepare_request(req)# ===== Step 3: 审计最终状态 =====print("="*60)print("PreparedRequest 审计")print("="*60)print(f"Method:{prepared.method}")print(f"URL:{prepared.url}")print(f"\nHeaders ({len(prepared.headers)}):")fork,vinprepared.headers.items():source=""ifkin("Host","User-Agent","Accept-Encoding","Accept","Connection"):source=" [隐式]"elifk=="X-API-Key":source=" [Session默认]"elifk=="X-Request-ID":source=" [请求显式]"elifk=="Content-Length":source=" [自动计算]"print(f"{k}:{v}{source}")print(f"\nBody:{prepared.body.decode()ifprepared.bodyelse'(空)'}")# ===== Step 4: 转为 curl 命令 =====defto_curl(p:requests.PreparedRequest)->str:parts=[f"curl -X{p.method}"]fork,vinp.headers.items():parts.append(f'-H "{k}:{v}"')ifp.body:parts.append(f"-d '{p.body.decode()}'")parts.append(f"'{p.url}'")return" \\\n ".join(parts)print(f"\nCurl 等价命令:")print(to_curl(prepared))运行输出（关键部分）：
============================================================ PreparedRequest 审计 ============================================================ Method: POST URL: https://httpbin.org/post?source=web Headers (8): X-API-Key: default-key [Session默认] X-Request-ID: req-001 [请求显式] Content-Type: application/json [自动设置] Content-Length: 44 [自动计算] User-Agent: python-requests/2.32.4 [隐式] Accept-Encoding: gzip, deflate [隐式] Accept: */* [隐式] Connection: keep-alive [隐式] Host: httpbin.org [隐式] Body: {"user": "\u5c0f\u80d6", "action": "test"}步骤二：批量发送——模板复用
目标：预编译请求模板，循环替换 body 后批量发送，对比编码开销。
importrequestsimporttimeimportjson# ===== 场景：向 100 个用户发送个性化通知 =====users=[{"id":i,"name":f"user_{i}"}foriinrange(100)]url="https://httpbin.org/post"# ----- 方式 A：每次调 requests.post() -----print("方式 A: requests.post() 每次构建")session=requests.Session()start=time.perf_counter()foruserinusers:resp=session.post(url,json=user,timeout=10)elapsed_a=time.perf_counter()-startprint(f" 耗时:{elapsed_a:.3f}s (100 次完整构建+发送)")# ----- 方式 B：PreparedRequest 模板复用 -----print(f"\n方式 B: PreparedRequest 模板复用")session=requests.Session()# 构建模板（只做一次）template=Request("POST",url,headers={"Content-Type":"application/json"})prepared=session.prepare_request(template)start=time.perf_counter()foruserinusers:# 只替换 bodyprepared.body=json.dumps(user).encode("utf-8")resp=session.send(prepared,timeout=10)elapsed_b=time.perf_counter()-startprint(f" 耗时:{elapsed_b:.3f}s (1 次预编译 + 100 次发送)")print(f"\n性能差异: 方式 B 比方式 A{'快'ifelapsed_b<elapsed_aelse'慢'}"f"{abs(elapsed_a-elapsed_b):.3f}s")print("注意：差异主要体现在构建开销上，网络时间二者相同")session.close()步骤三：PreparedRequest + 请求签名
目标：在 prepare 之后、send 之前插入 HMAC-SHA256 签名。
importrequestsimporthmacimporthashlibimporttimeimportjsonclassSignedAPIClient:"""带请求签名的 API 客户端"""def__init__(self,api_key:str,api_secret:str):self.api_key=api_key self.api_secret=api_secret.encode()self.session=requests.Session()self.session.headers["X-API-Key"]=self.api_keydefsigned_request(self,method:str,url:str,**kwargs)->requests.Response:"""发送签名请求"""# 1. 构建 Requestreq=requests.Request(method,url,**kwargs)# 2. 预编译prepared=self.session.prepare_request(req)# 3. 签名（在 prepare 之后、send 之前）self._sign(prepared)# 4. 发送returnself.session.send(prepared)def_sign(self,prepared:requests.PreparedRequest):"""对 PreparedRequest 进行 HMAC-SHA256 签名"""timestamp=str(int(time.time()))body=prepared.bodyorb""# 签名字符串：method + path + timestamp + bodysign_str=(f"{prepared.method}{prepared.path_url}{timestamp}").encode()+body signature=hmac.new(self.api_secret,sign_str,hashlib.sha256).hexdigest()# 将签名信息注入 headerprepared.headers["X-Timestamp"]=timestamp prepared.headers["X-Signature"]=signaturedefpost(self,url:str,**kwargs)->requests.Response:returnself.signed_request("POST",url,**kwargs)defget(self,url:str,**kwargs)->requests.Response:returnself.signed_request("GET",url,**kwargs)defclose(self):self.session.close()# ===== 使用演示 =====client=SignedAPIClient("my-api-key","my-super-secret")# 发送签名请求resp=client.post("https://httpbin.org/post",json={"action":"transfer","amount":100},)data=resp.json()print("服务端收到的签名头:")forkin["X-Api-Key","X-Timestamp","X-Signature"]:print(f"{k}:{data['headers'].get(k,'N/A')[:40]}...")client.close()可能遇到的坑及解决方法
坑1：PreparedRequest.body 是可变引用
# prepared.body 是一个 bytes 对象的引用# 虽然 PreparedRequest 本身是"不可变的"，但 body 可以被替换prepared.body=new_body# 这是故意支持的，方便模板复用坑2：修改 prepared.body 后 Content-Length 不会自动更新
# 如果你替换了 body，需要手动更新 Content-Length 头prepared.body=json.dumps(new_data).encode()prepared.headers["Content-Length"]=str(len(prepared.body))# 或使用 prepare_body 方法重新处理坑3：prepared.headers 是 CaseInsensitiveDict
# headers 的 key 大小写不敏感，但 value 是字符串prepared.headers["content-type"]# 等价于 prepared.headers["Content-Type"]# 但 Session.send() 后 headers 可能被再次修改测试验证
importpytestimportrequestsclassTestPreparedRequest:"""验证 PreparedRequest 功能"""deftest_prepare_basic_get(self):s=requests.Session()req=requests.Request("GET","https://httpbin.org/get",params={"q":"test"})prepared=s.prepare_request(req)assertprepared.method=="GET"assert"q=test"inprepared.url s.close()deftest_prepare_includes_implicit_headers(self):s=requests.Session()s.headers["X-Custom"]="session-level"req=requests.Request("GET","https://httpbin.org/get")prepared=s.prepare_request(req)assert"Host"inprepared.headersassertprepared.headers["X-Custom"]=="session-level"s.close()deftest_prepare_and_send(self):s=requests.Session()req=requests.Request("GET","https://httpbin.org/get")prepared=s.prepare_request(req)resp=s.send(prepared)assertresp.status_code==200s.close()deftest_template_reuse(self):s=requests.Session()req=requests.Request("POST","https://httpbin.org/post",headers={"Content-Type":"application/json"})template=s.prepare_request(req)importjsonforiinrange(3):template.body=json.dumps({"i":i}).encode()resp=s.send(template,timeout=10)assertresp.status_code==200assertresp.json()["json"]["i"]==i s.close()4. 项目总结
核心知识点
| 步骤 | API | 作用 |
|---|---|---|
| 构建 Request | Request(method, url, **kwargs) | 收集所有请求参数 |
| 预编译 | session.prepare_request(Request) | 执行编码/合并/序列化 |
| 审计 | prepared.url/headers/body | 查看最终请求内容 |
| 签名 | 修改prepared.headers | 插入签名/认证信息 |
| 发送 | session.send(prepared) | 通过 Session 发送 |
优点 & 缺点
| 对比维度 | requests.get() | PreparedRequest + send() |
|---|---|---|
| 批量场景性能 | 每次完整构建 | 模板复用 |
| 可见性（审计） | 不透明 | 完全透明 |
| 签名/认证窗口 | 不方便 | 黄金窗口 |
| 代码复杂度 | 极简 | 略高 |
适用场景
- 批量 API 调用（同模板不同参数）
- 请求签名（HMAC/AWS SigV4/OAuth）
- 安全审计（查看最终发出的完整请求）
- 调试（PreparedRequest 转 curl 命令）
注意事项
- 修改
prepared.body后需手动更新Content-Length Session.send()会再次处理prepared（如认证 hook），不是简单的"发送"- PreparedRequest 的
path_url不包含 host（只含 path + query）
常见踩坑经验
案例一：批量发送时 Content-Length 不更新。开发用模板发送 100 条消息，替换了 body 但忘了更新 Content-Length。前 50 条数据都正确（因为 body 长度恰好一样），第 51 条数据更长——服务端只读到了前 50 条数据的长度，数据被截断。根因：body 长度变化后 Content-Length 未更新。修复：每次替换 body 后prepared.headers["Content-Length"] = str(len(prepared.body))或调用prepared.prepare_body()。
案例二：PreparedRequest 的 headers 被 Session 再次修改。开发在 PreparedRequest 中正确设置了所有 headers，但Session.send()执行时 Session 的auth钩子修改了 Authorization 头。根因：Session.send() 不是纯粹的网络发送，它还会执行 Session 级别的 hooks 和 auth。修复：理解 Session.send() 的完整流程，在签名后设置不可覆盖的标记。
案例三：PreparedRequest 模板并发安全问题。批量发送使用了多线程，所有线程共享同一个 PreparedRequest 模板但各自替换 body——在并发环境下产生了竞态条件（线程 A 的 body 被线程 B 覆盖）。根因：PreparedRequest 不是线程安全的。修复：每个线程独立创建 PreparedRequest 实例，或使用deepcopy。
思考题
设计题：设计一个"请求重放"工具——将 PreparedRequest 序列化为 JSON/YAML 文件，支持在另一个环境中加载并重新发送。需要处理的问题：序列化 headers（CaseInsensitiveDict）、序列化 body（可能是 bytes）、反序列化后如何重新发送（Session 不存在了）。
扩展题：如何在 PreparedRequest 的基础上实现请求的"乐观重试"——如果预测到请求可能会失败（基于历史模式），在 prepare 阶段就生成一个"备选 PreparedRequest"（指向备用服务器或降级路径），并在主请求失败时自动切换到备选？
延伸阅读与资源
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析
