跳转至

Python HTTP 客户端实战(requests)

调用 HTTP 接口是脚本与服务最常见的工作。标准库 urllib 接口繁琐,事实标准是第三方库 requests(本篇在 requests 2.32 上、针对本地测试服务器全量验证)。环境安装见 Python 环境与包管理;自己对外提供 HTTP 服务用 Nginx 加应用框架。

安装与第一个请求

pip install requests        # 或 uv add requests
import requests

r = requests.get("https://httpbin.org/get", params={"q": "python", "page": 1})
print(r.status_code)        # 200
print(r.url)                # https://httpbin.org/get?q=python&page=1(params 自动拼接)
print(r.headers["Content-Type"])
print(r.text[:80])          # 响应体(文本)

params 自动做 URL 编码——不要手动拼查询字符串。

GET:参数、头与状态码

r = requests.get(
    "https://api.example.com/data",
    params={"id": 42},
    headers={"X-Token": "secret", "Accept": "application/json"},
    timeout=(3.05, 10),          # (连接超时, 读超时)——必须设,见下文
)

if r.status_code != 200:
    r.raise_for_status()         # 非 2xx 抛 HTTPError(带上状态码与 URL)
data = r.json()                  # 解析 JSON 响应(失败抛 requests.exceptions.JSONDecodeError)

状态码判断的两种习惯:if r.ok(2xx 为 True)分支处理,或 r.raise_for_status() 统一抛异常(配合 try/except,见错误处理思想)。

POST:JSON 与表单

# 发送 JSON(最常见):json= 自动设置 Content-Type: application/json 并序列化
r = requests.post("https://api.example.com/users",
                  json={"name": "ada", "score": 90},
                  timeout=10)

# 发送表单:data= 走 application/x-www-form-urlencoded
r = requests.post("https://api.example.com/login",
                  data={"user": "ada", "pass": "secret"},
                  timeout=10)

# 上传文件:files=
r = requests.post("https://api.example.com/upload",
                  files={"file": open("report.pdf", "rb")},
                  timeout=60)

三个参数的区别:json=(JSON 体)、data=(表单或原始体)、files=(multipart 上传),一次请求只用一个。

超时:唯一不许省的参数

requests.get(url, timeout=5)          # 连接+读共享 5 秒
requests.get(url, timeout=(3.05, 30)) # 连接 3 秒、读 30 秒(推荐元组写法)

不设 timeout 的 requests 请求可能永久挂起——连接建立了但服务器不吐数据时没有兜底。长轮询接口才需要放宽读超时,连接超时保持 3 秒左右不变。

Session:连接复用与公共配置

with requests.Session() as s:
    s.headers.update({"Authorization": "Bearer xxx"})   # 本会话所有请求带这个头
    s.mount("https://", adapter)                        # 挂重试策略(见下)

    r1 = s.get("https://api.example.com/users", timeout=10)
    r2 = s.get("https://api.example.com/orders", timeout=10)   # 复用连接池,快一截

同一主机的多个请求用 Session——TCP/TLS 握手只做一次。登录态(Cookie)也自动保持在 Session 里。

重试:只对幂等请求自动重试

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(total=3, backoff_factor=0.5,
              status_forcelist=[429, 500, 502, 503, 504],
              allowed_methods=["GET"])            # POST 默认不重试(可能重复下单)
adapter = HTTPAdapter(max_retries=retry)

s = requests.Session()
s.mount("https://", adapter)
s.mount("http://", adapter)

backoff_factor=0.5 的退避间隔约为 0.5s、1s、2s——避免把过载的服务打死。

下载大文件:流式

with requests.get(url, stream=True, timeout=30) as r:
    r.raise_for_status()
    with open("big.bin", "wb") as f:
        for chunk in r.iter_content(chunk_size=1 << 16):    # 64KB 一块
            f.write(chunk)

stream=True 时响应体不立即下载,iter_content 边收边写,内存占用恒定。用 with 包住响应确保连接释放。

错误处理全景

try:
    r = requests.get(url, timeout=5)
    r.raise_for_status()
except requests.exceptions.Timeout:
    ...                              # 超时(连接或读)
except requests.exceptions.HTTPError as e:
    ...                              # 4xx/5xx(raise_for_status 抛出)
except requests.exceptions.ConnectionError:
    ...                              # DNS 失败、拒绝连接等网络层问题
except requests.exceptions.RequestException as e:
    ...                              # 兜底:requests 所有异常的基类

异常继承链的末端是 RequestException,兜底捕获它即可不漏。

常见错误

错误 说明
不设 timeout 默认无限等待,服务慢时线程堆积——每个请求都要带
忽略状态码 r.json() 在错误页面上解析必然失败;先 raise_for_status() 或检查 r.ok
误用 data= 发 JSON 服务端收不到 JSON 体;发 JSON 用 json=
大文件 .text 全量载入 用 stream=True 分块
验证失败直接 -k/verify=False 关闭证书校验只是掩盖问题(且报 InsecureRequestWarning);自签名场景把自签 CA 装进信任链
爬虫频率过高 加 time.sleep 或限速;尊重 robots 与接口条款
敏感信息进日志 打日志前脱敏 headers/token

延伸:服务端与代理见 Nginx 反向代理入门;用 curl 在命令行快速验证接口见 curl 详解;把请求逻辑纳入测试用 mock(见测试与调试)。

评论