> ## Content Index > Fetch the complete content index at: https://blog.vercanti.com/llms.txt > Use this file to discover other available public pages before exploring further. # requests 完全指南 - URL: https://blog.vercanti.com/requests-wan-quan-zhi-nan/ - Published: 2026-08-28T14:34:47.000Z - Updated: 2026-08-28T14:57:20.000Z - Description: requests 是 Python 最流行的 HTTP 客户端库，封装了 urllib3，提供简洁的 API。适用于同步场景；异步场景推荐使用 httpx完全指南(/httpx-wan-quan-zhi-nan/)。 requests 提供对应各 HTTP 方法的函数，均返回 Response 对象。 所有请求方法共享以下参数（部分方法不适用的参数会被忽略）。 Session 对象在多次请求间复用 TCP 连接（HTTP Keep-Alive），并自动持久化 Cookie 和自定义 Header。 mount(prefix, adapter) 为特定 U - Author: yellowdog - Tags: Python, 框架与库 > 官方文档： > 适用版本：requests 2.32+（2026-05-07 核实） requests 是 Python 最流行的 HTTP 客户端库，封装了 urllib3，提供简洁的 API。适用于同步场景；异步场景推荐使用 [httpx完全指南](https://blog.vercanti.com/httpx-wan-quan-zhi-nan/)。 ## 安装 ```bash pip install requests ``` --- ## 基础请求方法 requests 提供对应各 HTTP 方法的函数，均返回 `Response` 对象。 ```python import requests resp = requests.get("https://httpbin.org/get") resp = requests.post("https://httpbin.org/post") resp = requests.put("https://httpbin.org/put") resp = requests.delete("https://httpbin.org/delete") resp = requests.patch("https://httpbin.org/patch") resp = requests.head("https://httpbin.org/get") resp = requests.options("https://httpbin.org/get") ``` ### 公共参数表格 所有请求方法共享以下参数（部分方法不适用的参数会被忽略）。 | 参数 | 类型 | 默认值 | 说明 | | ---------------- | ------------------------------ | -------------------- | -------------------------------------------------------------- | | url | str | — | 请求 URL | | params | dict / list / str | None | URL 查询参数，自动拼接到 URL 后；如 {"q": "python"} → ?q=python | | data | dict / list / bytes / str / IO | None | 请求体，以 application/x-www-form-urlencoded 发送（dict）或原始 bytes | | json | any | None | 请求体，自动序列化为 JSON 并设置 Content-Type: application/json | | headers | dict | None | 自定义请求头，与默认头合并 | | cookies | dict / CookieJar | None | 随请求发送的 Cookie | | auth | tuple / AuthBase | None | 认证凭据，如 ("user", "pass") 或 HTTPBasicAuth(...) | | timeout | float / tuple | None | 超时秒数；None 为永不超时；tuple (connect, read) 分别控制 | | proxies | dict | None | 代理，如 {"http": "http://host:port", "https": "http://host:port"} | | verify | bool / str | True | 是否验证 SSL 证书；可传 CA 证书路径 | | stream | bool | False | True 时不立即下载响应体，用于流式传输大文件 | | allow\_redirects | bool | True（GET）/ False（其他） | 是否自动跟随重定向 | | cert | str / tuple | None | 客户端证书路径或 (cert, key) 元组 | ```python # params 示例 resp = requests.get( "https://api.example.com/search", params={"q": "python", "page": 2, "tags": ["web", "scraping"]}, ) print(resp.url) # https://api.example.com/search?q=python&page=2&tags=web&tags=scraping # 发送 JSON resp = requests.post( "https://api.example.com/data", json={"key": "value"}, headers={"Authorization": "Bearer token123"}, timeout=10, ) # 使用代理 proxies = { "http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890", } resp = requests.get("https://example.com", proxies=proxies) ``` --- ## Response 对象 ### 属性与方法表格 | 属性/方法 | 类型 | 说明 | | -------------------------- | ------------------- | -------------------------------- | | status\_code | int | HTTP 状态码，如 200、404、500 | | headers | CaseInsensitiveDict | 响应头，键名大小写不敏感 | | text | str | 响应体解码后的字符串（编码由 encoding 决定） | | content | bytes | 响应体原始字节 | | json() | any | 将响应体解析为 JSON，失败抛 JSONDecodeError | | url | str | 最终请求的 URL（跟随重定向后） | | history | list\[Response\] | 重定向历史，每个元素为一个中间响应 | | encoding | str | 用于解码 text 的编码，可手动修改 | | apparent\_encoding | str | chardet 检测到的编码（比 encoding 更准确） | | elapsed | timedelta | 从发送请求到收到响应头的耗时 | | ok | bool | status\_code < 400 时为 True | | reason | str | 状态原因短语，如 "OK"、"Not Found" | | cookies | RequestsCookieJar | 响应中设置的 Cookie | | raise\_for\_status() | — | 状态码 >= 400 时抛 HTTPError | | iter\_content(chunk\_size) | generator | 流式迭代响应体（需 stream=True） | | iter\_lines() | generator | 按行迭代响应体（需 stream=True） | ```python resp = requests.get("https://httpbin.org/get", timeout=10) # 检查状态 resp.raise_for_status() # 非 2xx/3xx 抛异常 # 解析 JSON data = resp.json() # 修正编码后获取文本 resp.encoding = resp.apparent_encoding print(resp.text) # 查看耗时 print(resp.elapsed.total_seconds()) ``` --- ## Session `Session` 对象在多次请求间复用 TCP 连接（HTTP Keep-Alive），并自动持久化 Cookie 和自定义 Header。 ```python import requests session = requests.Session() # 设置会话级别的 headers 和 cookies（所有请求都会携带） session.headers.update({ "User-Agent": "Mozilla/5.0", "Accept-Language": "zh-CN,zh;q=0.9", }) session.cookies.set("session_id", "abc123") # 登录，session 自动保存响应中的 Set-Cookie session.post("https://example.com/login", data={"user": "foo", "pass": "bar"}) # 后续请求自动携带登录 Cookie resp = session.get("https://example.com/profile") # 使用完毕后关闭（释放连接池） session.close() # 推荐用上下文管理器 with requests.Session() as session: session.headers["Authorization"] = "Bearer token" resp = session.get("https://api.example.com/data") ``` ### session.mount() 挂载适配器 `mount(prefix, adapter)` 为特定 URL 前缀绑定自定义传输适配器。 | 参数 | 类型 | 默认值 | 说明 | | ------- | ----------- | --- | --------------------- | | prefix | str | — | URL 前缀，如 "https://" | | adapter | BaseAdapter | — | 适配器实例，通常为 HTTPAdapter | ### HTTPAdapter 参数 | 参数 | 类型 | 默认值 | 说明 | | ----------------- | ----------- | ----- | ------------------------------------- | | max\_retries | int / Retry | 0 | 最大重试次数，建议传 urllib3.util.Retry 对象以精细控制 | | pool\_connections | int | 10 | 连接池数量（对应不同主机的连接池个数） | | pool\_maxsize | int | 10 | 每个连接池最大连接数（并发请求数上限） | | pool\_block | bool | False | 连接池满时是否阻塞等待，False 时抛异常 | ```python from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry retry_strategy = Retry( total=3, # 最大重试次数 backoff_factor=1, # 退避系数：1 → 0s, 2s, 4s status_forcelist=[429, 500, 502, 503, 504], # 触发重试的状态码 allowed_methods=["GET", "POST"], ) adapter = HTTPAdapter( max_retries=retry_strategy, pool_connections=20, pool_maxsize=50, ) session = requests.Session() session.mount("https://", adapter) session.mount("http://", adapter) ``` --- ## 认证 ### HTTPBasicAuth 与 HTTPDigestAuth ```python from requests.auth import HTTPBasicAuth, HTTPDigestAuth # Basic Auth（明文 base64 编码，须配合 HTTPS） resp = requests.get( "https://api.example.com/private", auth=HTTPBasicAuth("username", "password"), ) # 等价简写 resp = requests.get("https://api.example.com/private", auth=("username", "password")) # Digest Auth（挑战-响应机制，安全性高于 Basic） resp = requests.get( "https://api.example.com/private", auth=HTTPDigestAuth("username", "password"), ) ``` ### Bearer Token ```python token = "eyJhbGci..." # 方式一：手动设置 Header resp = requests.get( "https://api.example.com/data", headers={"Authorization": f"Bearer {token}"}, ) # 方式二：Session 级别 session = requests.Session() session.headers["Authorization"] = f"Bearer {token}" ``` ### 自定义 AuthBase ```python from requests.auth import AuthBase class APIKeyAuth(AuthBase): def __init__(self, api_key): self.api_key = api_key def __call__(self, r): # r 是 PreparedRequest 对象 r.headers["X-API-Key"] = self.api_key return r resp = requests.get("https://api.example.com/data", auth=APIKeyAuth("my-key")) ``` --- ## 文件上传与下载 ### 文件上传 `files` 参数的格式决定了 Content-Type 为 `multipart/form-data`。 ```python # 单文件上传（最简形式） with open("photo.jpg", "rb") as f: resp = requests.post("https://api.example.com/upload", files={"file": f}) # 指定文件名和 MIME 类型 with open("report.pdf", "rb") as f: resp = requests.post( "https://api.example.com/upload", files={ "file": ("custom_name.pdf", f, "application/pdf"), }, ) # 同时上传文件和表单字段 with open("image.png", "rb") as f: resp = requests.post( "https://api.example.com/upload", files={"image": ("image.png", f, "image/png")}, data={"title": "My Image", "description": "Test"}, ) # 多文件上传 files = [ ("files", ("a.txt", open("a.txt", "rb"), "text/plain")), ("files", ("b.txt", open("b.txt", "rb"), "text/plain")), ] resp = requests.post("https://api.example.com/batch", files=files) ``` ### 流式下载大文件 下载大文件时必须使用 `stream=True`，否则整个响应体会被加载到内存中。 | 参数 | 类型 | 默认值 | 说明 | | --------------- | ---------- | ---- | ----------------------- | | chunk\_size | int / None | 1 | 每次读取的字节数；None 表示收到多少读多少 | | decode\_content | bool | True | 是否自动解压 gzip/deflate | ```python import requests url = "https://example.com/large_file.zip" with requests.get(url, stream=True, timeout=30) as resp: resp.raise_for_status() total = int(resp.headers.get("Content-Length", 0)) downloaded = 0 with open("large_file.zip", "wb") as f: for chunk in resp.iter_content(chunk_size=8192): if chunk: # 过滤保持连接的空 chunk f.write(chunk) downloaded += len(chunk) if total: pct = downloaded / total * 100 print(f"\r下载进度: {pct:.1f}%", end="") ``` --- ## 高级用法 ### 自定义重试策略 详细参数见上方 `HTTPAdapter` 部分，以下补充 `Retry` 的完整参数。 | 参数 | 类型 | 默认值 | 说明 | | ----------------- | ----- | --------------------------------- | -------------------------------------------------------- | | total | int | 10 | 总重试次数 | | connect | int | None | 连接错误重试次数 | | read | int | None | 读取错误重试次数 | | redirect | int | None | 重定向次数上限 | | status | int | None | 按状态码触发重试的次数 | | status\_forcelist | set | None | 触发重试的 HTTP 状态码集合 | | allowed\_methods | set | frozenset(\["GET","HEAD","..."\]) | 允许重试的请求方法 | | backoff\_factor | float | 0 | 退避系数，等待时间 = backoff\_factor \* (2 \*\* (retry\_num - 1)) | | raise\_on\_status | bool | False | 超过重试次数后是否抛 MaxRetryError | ### 请求与响应 Hook Hook 在请求/响应周期的特定时机被调用。目前仅支持 `response` 事件。 ```python import requests def log_response(resp, *args, **kwargs): print(f"[{resp.status_code}] {resp.url} ({resp.elapsed.total_seconds():.3f}s)") def check_rate_limit(resp, *args, **kwargs): if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After", 60)) print(f"触发限流，等待 {retry_after} 秒") import time time.sleep(retry_after) # 重试 return resp.connection.send(resp.request) # 单次请求 hook resp = requests.get( "https://api.example.com/data", hooks={"response": [log_response, check_rate_limit]}, ) # Session 级别 hook（对所有请求生效） session = requests.Session() session.hooks["response"].append(log_response) ``` ### PreparedRequest 预构建请求 `PreparedRequest` 允许在发送前检查或修改完整的请求对象（URL、头、体）。 ```python from requests import Request, Session req = Request( method="POST", url="https://api.example.com/data", headers={"X-Custom": "value"}, json={"key": "val"}, ) session = Session() prepared = session.prepare_request(req) # 检查最终请求内容 print(prepared.url) print(prepared.headers) print(prepared.body) # 发送 resp = session.send(prepared, timeout=10) ``` ### 超时细粒度控制 `timeout` 参数接受 tuple `(connect_timeout, read_timeout)`： | 超时类型 | 含义 | | --------------------- | -------------------------- | | 连接超时（connect timeout） | 建立 TCP 连接的最长等待时间 | | 读取超时（read timeout） | 等待服务端发送响应数据的最长间隔时间（非总传输时间） | ```python # 连接超时 5 秒，读取超时 30 秒 resp = requests.get("https://example.com", timeout=(5, 30)) # 所有阶段统一 10 秒 resp = requests.get("https://example.com", timeout=10) # 永不超时（生产环境不推荐） resp = requests.get("https://example.com", timeout=None) ``` --- ## 最佳实践 ### 统一封装请求客户端 ```python import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session( retries: int = 3, backoff_factor: float = 0.5, pool_maxsize: int = 20, headers: dict = None, ) -> requests.Session: session = requests.Session() retry = Retry( total=retries, backoff_factor=backoff_factor, status_forcelist={429, 500, 502, 503, 504}, allowed_methods={"GET", "POST"}, ) adapter = HTTPAdapter(max_retries=retry, pool_maxsize=pool_maxsize) session.mount("https://", adapter) session.mount("http://", adapter) if headers: session.headers.update(headers) return session client = build_session(headers={"User-Agent": "MyCrawler/1.0"}) resp = client.get("https://example.com", timeout=(5, 15)) resp.raise_for_status() ``` ### 安全检查响应 ```python def safe_get(session, url, **kwargs): try: resp = session.get(url, timeout=(5, 15), **kwargs) resp.raise_for_status() return resp except requests.exceptions.Timeout: print(f"请求超时: {url}") except requests.exceptions.ConnectionError: print(f"连接失败: {url}") except requests.exceptions.HTTPError as e: print(f"HTTP 错误 {e.response.status_code}: {url}") return None ``` --- ## 踩坑与注意事项 ### verify=False 的安全问题 ```python # 危险：跳过 SSL 证书验证，中间人攻击无法被检测 resp = requests.get("https://example.com", verify=False) ``` `verify=False` 会同时触发 `InsecureRequestWarning`。生产环境绝不应关闭验证。正确做法： 1. 升级 `certifi`：`pip install -U certifi` 2. 传入自签名 CA 证书路径：`verify="/path/to/ca-bundle.crt"` 3. 设置系统信任的 CA 证书：`export REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt` ### 连接池耗尽 症状：并发请求时出现 `urllib3.exceptions.MaxRetryError: ... pool is full`。 原因：`pool_maxsize` 默认为 `10`，并发超过上限时多余连接被丢弃或阻塞。 ```python # 解决：根据并发数调整 pool_maxsize adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50) session.mount("https://", adapter) ``` 多线程场景下每个线程共享同一个 Session 实例即可复用连接池；若每个线程各创建 Session 则完全失去复用效果。 ### 大文件内存溢出 直接使用 `resp.content` 或 `resp.text` 会将整个响应体加载到内存，下载大文件时会 OOM。 ```python # 错误：内存溢出风险 content = requests.get(large_url).content with open("file", "wb") as f: f.write(content) # 正确：流式写入 with requests.get(large_url, stream=True) as resp: with open("file", "wb") as f: for chunk in resp.iter_content(chunk_size=65536): f.write(chunk) ``` ### data 与 json 参数的 Content-Type 区别 | 参数 | Content-Type | 适用场景 | | --------------- | --------------------------------- | ------------------ | | data={"k": "v"} | application/x-www-form-urlencoded | 传统 HTML 表单提交 | | data=raw\_bytes | 无自动设置，取 headers 中的值 | 发送原始数据 | | json={"k": "v"} | application/json | REST API、现代 Web 接口 | 同时传 `data` 和 `json` 时，`json` 参数优先，`data` 被忽略。 ```python # 接口要求 JSON 但误用了 data（服务端会按表单解析，通常报错） resp = requests.post(url, data={"key": "value"}) # 错误 # 正确：使用 json 参数 resp = requests.post(url, json={"key": "value"}) # 正确 ``` ### requests 不支持异步 requests 是同步阻塞库，无法在 `asyncio` 事件循环中直接使用。异步场景请使用 [httpx完全指南](https://blog.vercanti.com/httpx-wan-quan-zhi-nan/)（API 与 requests 高度兼容）或 `aiohttp`。 ```python # 错误：在 async 函数中使用 requests 会阻塞事件循环 async def fetch(url): return requests.get(url) # 阻塞整个事件循环 # 正确：使用 httpx import httpx async def fetch(url): async with httpx.AsyncClient() as client: return await client.get(url) ``` 详见 [httpx完全指南](https://blog.vercanti.com/httpx-wan-quan-zhi-nan/)。 --- ## 常见陷阱 ### 陷阱：未使用 `Session` 导致频繁建立 TCP 连接 **现象：** 高频请求同一域名时性能差，每次都看到新 TCP 握手。 **原因：** 直接调用 `requests.get()` 每次创建临时 Session，不复用连接池，TCP 连接无法复用。 **解决：** 批量请求同一服务时使用 `requests.Session()`，自动复用 Keep-Alive 连接： ```python with requests.Session() as session: session.headers.update({'Authorization': f'Bearer {token}'}) for url in urls: resp = session.get(url, timeout=10) ``` ### 陷阱：未设置 `timeout` 导致请求永久挂起 **现象：** 服务端无响应时，程序卡住，等待无限长时间。 **原因：** `requests` 默认无超时，若服务端不关闭连接，`get()` 会一直等待。 **解决：** 始终设置 `timeout=(connect_timeout, read_timeout)` 元组： ```python resp = requests.get(url, timeout=(3.05, 30)) # 连接超时 3.05s，读取超时 30s ``` ### 陷阱：大文件下载将内容全部加载到内存 **现象：** 下载大文件时内存暴增，甚至 OOM。 **原因：** `resp.content` 或 `resp.text` 将整个响应体读入内存，不适合大文件。 **解决：** 使用 `stream=True` 配合分块写入： ```python with requests.get(url, stream=True, timeout=30) as resp: with open('file.bin', 'wb') as f: for chunk in resp.iter_content(chunk_size=8192): f.write(chunk) ``` --- ## 参见 [httpx完全指南](https://blog.vercanti.com/httpx-wan-quan-zhi-nan/) [Celery完全指南](https://blog.vercanti.com/celery-wan-quan-zhi-nan/)