1. 引言
requests 是 Python 生态中最常用的 HTTP 客户端库。初学者通常只掌握get、post和简单的参数传递，但在真实项目中，我们还需要处理连接复用、自动重试、认证、代理、文件上传、异常恢复和并发请求等问题。本文围绕这些进阶能力展开，帮助你写出更稳定、更高效的网络请求代码。
2. Session：连接复用与状态保持
如果每次请求都调用顶层函数requests.get，底层会为每个请求重新建立 TCP 连接，Cookie 也不会自动保存。使用Session可以复用底层连接池，并在同一会话内保持 Cookie，从而减少握手开销、模拟真实用户登录状态。
import requests session = requests.Session() session.headers.update({"User-Agent": "my-app/1.0"}) 设置 Cookie 后在同一会话内自动携带 session.get("https://httpbin.org/cookies/set/name/value") resp = session.get("https://httpbin.org/cookies") print(resp.json())需要特别注意的是，不要在Session和顶层函数之间混用状态。登录、下单这类连续操作应始终通过同一个Session实例完成。
3. 超时与自动重试：让请求更健壮
网络请求可能出现连接超时、读取超时或服务端临时错误。默认情况下 requests 没有超时限制，也没有重试机制。推荐为每次请求显式设置timeout，并借助Retry和HTTPAdapter实现自动重试。
import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retry = Retry( total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504], allowed_methods=["GET", "POST"], ) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) 服务端返回 503 时会按退避策略自动重试 resp = session.get("https://example.com", timeout=5)total控制最大重试次数，backoff_factor控制退避间隔，status_forcelist指定哪些状态码需要重试。读写超时可以用元组分别设置，例如timeout=(3, 10)表示连接超时 3 秒、读取超时 10 秒。
4. 流式请求与文件下载
下载大文件时，如果使用resp.content会把整个文件一次性读入内存，容易导致内存暴涨。开启stream=True后，可以按块读取响应体并写入磁盘。
url = "https://example.com/large-file.zip" with requests.get(url, stream=True, timeout=30) as resp: resp.raise_for_status() with open("large-file.zip", "wb") as f: for chunk in resp.iter_content(chunk_size=8192): f.write(chunk)使用iter_content时建议显式指定chunk_size，否则部分实现会逐字节返回，影响效率。下载后还要通过raise_for_status检查 HTTP 状态码，避免把错误页面写入文件。
5. 认证机制：Basic、Token 与自定义认证
requests 内置了多种认证方式。对于 Basic 认证或 Digest 认证，可以直接传入认证对象；对于常见的 Bearer Token，可以自定义认证类。
from requests.auth import HTTPBasicAuth, HTTPDigestAuth, AuthBase 内置认证 r1 = requests.get(url, auth=HTTPBasicAuth("user", "pass")) r2 = requests.get(url, auth=HTTPDigestAuth("user", "pass")) 自定义 Bearer Token 认证 class TokenAuth(AuthBase): def init(self, token): self.token = token def __call__(self, request): request.headers["Authorization"] = f"Bearer {self.token}" return request r3 = requests.get(url, auth=TokenAuth("your-token"))自定义认证类需要继承AuthBase并实现__call__方法，在方法内修改请求对象后返回。这样可以灵活对接企业内部的签名认证逻辑。
6. 代理与 TLS/SSL 配置
爬虫、外网访问或公司内网环境经常需要配置代理。requests 通过proxies参数支持 HTTP、HTTPS 代理，并通过verify控制证书校验。
proxies = { "http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890", } 使用代理并指定自定义 CA 证书 resp = requests.get( "https://example.com", proxies=proxies, verify="/path/to/ca-bundle.pem", timeout=10, )如果请求目标是自签名证书或测试环境，可以临时设置verify=False，但这会带来安全风险，生产环境应优先提供正确的 CA 证书。客户端证书可通过cert参数传入。
7. 钩子机制：拦截请求与响应
钩子允许我们在响应返回后的特定时机执行自定义逻辑，例如统一记录日志、打印耗时或触发告警。
def log_response(resp, *args, **kwargs): print(f"request url: {resp.url}") print(f"status code: {resp.status_code}") resp = requests.get( "https://example.com", hooks={"response": log_response}, timeout=10, )钩子函数必须接受响应对象作为第一个参数，后续还可以接收其他关键字参数。多个钩子可以按列表传入，requests 会依次调用。基于钩子可以在不侵入业务代码的前提下统一处理响应。
8. 文件上传与多部分表单
上传文件时，requests 会自动构造multipart/form-data请求。通过files参数可以同时上传多个文件，并为每个文件指定文件名和 MIME 类型。
with open("report.csv", "rb") as csv_file: files = { "file": ("report.csv", csv_file, "text/csv"), "attachment": ("note.txt", open("note.txt", "rb"), "text/plain"), } data = {"category": "finace"} resp = requests.post( "https://example.com/upload", files=files, data=data, timeout=30, )元组中的三个元素分别表示文件名、文件对象和 MIME 类型。对于大文件，建议使用with打开并保持流式处理，避免一次性读入内存。服务端返回后应检查状态码，并根据业务约定判断上传是否成功。
9. 错误处理与异常体系
requests 的异常都继承自RequestException。编写健壮的请求代码时，应捕获具体异常并区分超时、连接错误和 HTTP 错误，而不是用宽泛的except吞掉所有问题。
import requests try: resp = requests.get("https://example.com", timeout=5) resp.raise_for_status() data = resp.json() except requests.exceptions.Timeout: print("请求超时") except requests.exceptions.ConnectionError: print("连接失败") except requests.exceptions.HTTPError as exc: print(f"HTTP 错误: {exc}") except requests.exceptions.RequestException as exc: print(f"其他请求异常: {exc}")raise_for_status只对 4xx 和 5xx 状态码抛出HTTPError，不会替我们处理网络层错误。建议把超时、连接错误和 HTTP 错误分开处理，并根据业务需要决定是否重试或降级。
10. 并发请求提升吞吐
requests 本身是同步阻塞的，串行请求大量接口时耗时较长。利用线程池可以显著提升 IO 密集型场景的吞吐量。每个线程持有独立连接，CPU 开销较低，适合爬虫和批量数据拉取。
from concurrent.futures import ThreadPoolExecutor, as_completed urls = [ "https://example.com/api/1", "https://example.com/api/2", "https://example.com/api/3", ] def fetch(url): return requests.get(url, timeout=10).json() with ThreadPoolExecutor(max_workers=8) as executor: future_to_url = {executor.submit(fetch, url): url for url in urls} for future in as_completed(future_to_url): url = future_to_url[future] try: data = future.result() print(f"{url} -> {data}") except requests.exceptions.RequestException as exc: print(f"{url} failed: {exc}")当单机线程池仍不够用时，可以考虑异步方案。requests 底层依赖同步的 urllib3，若要真正异步，可以结合asyncio使用httpx，但 requests 配合线程池仍然是兼顾学习成本和性能的常用做法。
11. 总结
requests 的进阶能力主要围绕稳定性、可维护性和性能展开：用Session复用连接和 Cookie；用Retry与timeout应对网络抖动；用流式请求处理大文件；用自定义认证适配企业接口；用钩子统一处理响应；用异常体系分类处理错误；用线程池提升并发吞吐。掌握这些技巧后，你可以把 requests 从简单的“能发请求”提升为可靠的生产级 HTTP 客户端。
