requests 完全指南

官方文档：https://requests.readthedocs.io/en/latest/
适用版本：requests 2.32+（2026-05-07 核实）

requests 是 Python 最流行的 HTTP 客户端库，封装了 urllib3，提供简洁的 API。适用于同步场景；异步场景推荐使用 httpx完全指南。

安装

pip install requests

基础请求方法

requests 提供对应各 HTTP 方法的函数，均返回 Response 对象。

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")

公共参数表格

所有请求方法共享以下参数（部分方法不适用的参数会被忽略）。

# 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 对象

属性与方法表格

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。

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 前缀绑定自定义传输适配器。

HTTPAdapter 参数

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

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

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

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。

# 单文件上传（最简形式）
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，否则整个响应体会被加载到内存中。

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 的完整参数。

请求与响应 Hook

Hook 在请求/响应周期的特定时机被调用。目前仅支持 response 事件。

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、头、体）。

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)：

# 连接超时 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)

最佳实践

统一封装请求客户端

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()

安全检查响应

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 的安全问题

# 危险：跳过 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，并发超过上限时多余连接被丢弃或阻塞。

# 解决：根据并发数调整 pool_maxsize
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50)
session.mount("https://", adapter)

多线程场景下每个线程共享同一个 Session 实例即可复用连接池；若每个线程各创建 Session 则完全失去复用效果。

大文件内存溢出

直接使用 resp.content 或 resp.text 会将整个响应体加载到内存，下载大文件时会 OOM。

# 错误：内存溢出风险
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 区别

同时传 data 和 json 时，json 参数优先，data 被忽略。

# 接口要求 JSON 但误用了 data（服务端会按表单解析，通常报错）
resp = requests.post(url, data={"key": "value"})   # 错误

# 正确：使用 json 参数
resp = requests.post(url, json={"key": "value"})   # 正确

requests 不支持异步

requests 是同步阻塞库，无法在 asyncio 事件循环中直接使用。异步场景请使用 httpx完全指南（API 与 requests 高度兼容）或 aiohttp。

# 错误：在 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完全指南。

常见陷阱

陷阱：未使用 Session 导致频繁建立 TCP 连接

现象： 高频请求同一域名时性能差，每次都看到新 TCP 握手。
原因： 直接调用 requests.get() 每次创建临时 Session，不复用连接池，TCP 连接无法复用。
解决： 批量请求同一服务时使用 requests.Session()，自动复用 Keep-Alive 连接：

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) 元组：

resp = requests.get(url, timeout=(3.05, 30))
# 连接超时 3.05s，读取超时 30s

陷阱：大文件下载将内容全部加载到内存

现象： 下载大文件时内存暴增，甚至 OOM。
原因： resp.content 或 resp.text 将整个响应体读入内存，不适合大文件。
解决： 使用 stream=True 配合分块写入：

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完全指南
Celery完全指南