requests相关

Python requests 库 接口自动化 完整系统学习大纲

本文面向测试工程师。Postman 适合手工调试，但要写接口自动化框架，requests 是 Python 领域无可争议的标准库——它有 Postman 的全部能力，还能被 pytest 调度、被 Jenkins 定时执行、被数据驱动批量运行。 环境：Python 3.8+ / requests 2.31+

一、requests 概述

1.1 什么是 requests？

• 定义：requests 是 Python 的第三方 HTTP 客户端库，封装了 Python 标准库 urllib3，用极简的 API 完成 HTTP 请求。
• 作者：Kenneth Reitz，口号是 “HTTP for Humans”（让 HTTP 变简单）。
• 地位：Python 生态中下载量最高的包之一，几乎所有接口自动化框架（HttpRunner、JMeter 的 Python 扩展、各类自研平台）底层都用它。

1.2 为什么用 requests（对比 urllib）

# 标准库 urllib：发一个带 Header 的 POST，需要 5 步
import urllib.request, urllib.parse, json

data = json.dumps({"name": "zhangsan"}).encode()
req = urllib.request.Request(
    "https://api.example.com/user",
    data=data,
    headers={"Content-Type": "application/json"},
    method="POST"
)
with urllib.request.urlopen(req) as resp:
    body = json.loads(resp.read().decode())

# requests：一行搞定
import requests
body = requests.post("https://api.example.com/user", json={"name": "zhangsan"}).json()

维度	urllib	requests
代码量	繁琐（编码、构造 Request）	极简
Cookie 管理	需 http.cookiejar	Session 自动管理
JSON 处理	需手动 json.dumps/loads	json= / .json()
连接池	无	内置（urllib3）
文件上传	复杂	files= 一个参数
认证	需手写 Header	auth=

1.3 安装

pip install requests

# 国内加速
pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple

# 验证
python -c "import requests; print(requests.__version__)"

1.4 第一个请求

import requests

r = requests.get("https://httpbin.org/get")

print(r.status_code)      # 200
print(r.text)             # 响应体（字符串）
print(r.json())           # 响应体（自动转 dict）

二、HTTP 方法与参数

2.1 方法总览

import requests

requests.get(url, params=None, **kwargs)          # 查询
requests.post(url, data=None, json=None, **kwargs) # 新增
requests.put(url, data=None, **kwargs)             # 整体更新
requests.patch(url, data=None, **kwargs)           # 局部更新
requests.delete(url, **kwargs)                     # 删除
requests.head(url, **kwargs)                       # 只要响应头
requests.options(url, **kwargs)                    # 查询支持的方法

# 通用写法（方法动态传入时用它）
requests.request("GET", url, **kwargs)

2.2 GET 请求与查询参数

# 方式一：params 字典（推荐，自动 URL 编码）
params = {"page": 1, "size": 20, "keyword": "手机 壳"}
r = requests.get("https://api.example.com/products", params=params)
print(r.url)
# https://api.example.com/products?page=1&size=20&keyword=%E6%89%8B%E6%9C%BA+%E5%A3%B3

# 方式二：直接拼在 URL 上（需自己处理编码）
r = requests.get("https://api.example.com/products?page=1&size=20")

💡 params 会自动做 URL 编码，处理中文、空格、特殊字符时务必用 params，不要手拼。

2.3 POST 请求：data vs json（高频易错点）

参数	Content-Type	数据格式	服务端接收方式
data=dict	application/x-www-form-urlencoded	表单 a=1&b=2	request.form / @RequestParam
data=json_str	需手动指定 application/json	JSON 字符串	request.json / @RequestBody
json=dict	自动设为 application/json	JSON	request.json / @RequestBody

# 1. 表单提交（form 表单）
r = requests.post(url, data={"username": "test01", "password": "123456"})

# 2. JSON 提交（REST API 最常用）★ 推荐
r = requests.post(url, json={"username": "test01", "password": "123456"})

# 3. 手动序列化（不推荐，容易忘写 header）
import json
r = requests.post(
    url,
    data=json.dumps({"a": 1}, ensure_ascii=False).encode("utf-8"),
    headers={"Content-Type": "application/json"}
)

⚠️ 接口 400/415 常见原因：该用 json= 却用了 data=（或反之），导致服务端解析不到 body。抓包对比 Content-Type 即可定位（呼应抓包工具篇）。

2.4 文件上传

# 单文件
files = {"file": open("test.png", "rb")}
r = requests.post("https://api.example.com/upload", files=files)

# 指定文件名与类型
files = {"file": ("report.xlsx", open("report.xlsx", "rb"),
                  "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
r = requests.post(url, files=files)

# 多文件 + 附带表单字段
files = [
    ("images", ("a.png", open("a.png", "rb"), "image/png")),
    ("images", ("b.png", open("b.png", "rb"), "image/png")),
]
r = requests.post(url, files=files, data={"album": "test"})

# 用完记得关闭（或用 with）
with open("test.png", "rb") as f:
    r = requests.post(url, files={"file": f})

三、请求定制

3.1 请求头 Headers

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Content-Type": "application/json;charset=utf-8",
    "Accept": "application/json",
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIs...",
}
r = requests.get(url, headers=headers)

3.2 Cookie

# 方式一：直接传字典
cookies = {"sessionid": "abc123", "token": "xyz"}
r = requests.get(url, cookies=cookies)

# 方式二：RequestsCookieJar（从上次响应里拿）
r1 = requests.get(login_url)
r2 = requests.get(url, cookies=r1.cookies)

# 读取响应中的 Cookie
print(r.cookies.get_dict())

3.3 认证方式

# 1. Basic Auth
from requests.auth import HTTPBasicAuth
r = requests.get(url, auth=HTTPBasicAuth("user", "pass"))
r = requests.get(url, auth=("user", "pass"))   # 简写

# 2. Bearer Token（最常见）
headers = {"Authorization": f"Bearer {token}"}
r = requests.get(url, headers=headers)

# 3. Digest Auth
from requests.auth import HTTPDigestAuth
r = requests.get(url, auth=HTTPDigestAuth("user", "pass"))

# 4. 自定义签名类（见第九章）

3.4 超时 timeout（必须设置！）

# 单一超时（连接+读取共 5 秒）
r = requests.get(url, timeout=5)

# 分别设置：(连接超时, 读取超时) ★ 推荐
r = requests.get(url, timeout=(3, 27))

⚠️ requests 默认没有超时！生产/自动化脚本不设 timeout，接口挂死会让整个线程永久阻塞。这是新手最常见的坑。

3.5 代理 proxies

proxies = {
    "http":  "http://127.0.0.1:8888",
    "https": "http://127.0.0.1:8888",
}
r = requests.get(url, proxies=proxies, verify=False)

# 走 Charles 抓包（配合抓包工具篇调试自己的脚本）
# 需要账号密码的代理
proxies = {"http": "http://user:pass@10.0.0.1:8080"}

3.6 重定向

r = requests.get(url, allow_redirects=False)   # 禁止自动跳转
print(r.status_code)      # 301/302
print(r.headers["Location"])  # 跳转目标

r = requests.get(url)     # 默认 allow_redirects=True（自动跟随）
print(r.history)          # 跳转历史列表

3.7 SSL 证书

r = requests.get(url, verify=False)      # 跳过证书校验（测试环境用）
# 会输出 InsecureRequestWarning，可关闭
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

r = requests.get(url, verify="/path/ca-bundle.crt")  # 指定 CA 证书
r = requests.get(url, cert=("/path/client.crt", "/path/client.key"))  # 双向认证

3.8 流式请求 stream

# 大文件下载：不一次性读入内存
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    with open("big.zip", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)

# 逐行读取（日志流、SSE）
r = requests.get(url, stream=True)
for line in r.iter_lines():
    if line:
        print(line.decode("utf-8"))

四、响应处理

4.1 Response 对象属性

属性	类型	说明
status_code	int	状态码（200、404、500…）
ok	bool	状态码 < 400 为 True
reason	str	状态描述（”OK”、”Not Found”）
headers	dict	响应头（大小写不敏感）
text	str	响应体（字符串，按 encoding 解码）
content	bytes	响应体（字节，图片/文件用）
json()	dict/list	反序列化 JSON
encoding	str	当前编码（可手动改）
cookies	CookieJar	响应 Cookie
url	str	最终 URL（含跳转后）
history	list	重定向历史
elapsed	timedelta	请求耗时（性能测试常用）
request	PreparedRequest	对应的请求对象

r = requests.get(url)

r.status_code                  # 200
r.ok                           # True
r.elapsed.total_seconds()      # 0.352（响应耗时，秒）
r.headers["Content-Type"]      # application/json
r.headers.get("Content-Type")  # 推荐用 get，避免 KeyError

4.2 状态码判断

# 方式一：手动判断
if r.status_code == 200:
    ...

# 方式二：raise_for_status()（非 2xx 抛 HTTPError）★ 推荐
r = requests.get(url)
r.raise_for_status()
data = r.json()

# 方式三：断言（测试中最常用）
assert r.status_code == 200, f"预期 200，实际 {r.status_code}，响应：{r.text}"

4.3 编码问题（中文乱码）

r = requests.get(url)
print(r.encoding)          # requests 猜测的编码（可能不对）

r.encoding = "utf-8"       # 方式一：手动指定
print(r.text)

print(r.apparent_encoding)  # chardet 检测的编码
r.encoding = r.apparent_encoding

content = r.content.decode("utf-8")   # 方式二：字节自行解码（最可靠）

⚠️ 中文乱码：先 print(r.apparent_encoding) 看真实编码，再手动赋值 r.encoding。

4.4 JSON 解析与容错

import json
from requests.exceptions import JSONDecodeError

try:
    data = r.json()
except JSONDecodeError:
    print("响应不是合法 JSON：", r.text[:200])
    data = None

# 安全取值（避免 KeyError）
code = data.get("code")
items = data.get("data", {}).get("list", [])

五、会话保持：Session（重点）

5.1 为什么用 Session

requests.get() 每次都是独立连接：新建 TCP、新建 TLS 握手，Cookie 也不共享。 Session 会复用连接（连接池），并自动携带 Cookie，是登录态场景的必需品。

import requests

s = requests.Session()

# 1. 登录（Session 自动保存响应 Cookie）
login = s.post("https://api.example.com/login",
               json={"username": "test01", "password": "123456"})
print(s.cookies.get_dict())

# 2. 后续请求自动带上 Cookie，无需手动传
profile = s.get("https://api.example.com/profile")
orders = s.get("https://api.example.com/orders")

s.close()   # 用完关闭（或用 with）

5.2 Session 的复用配置

s = requests.Session()

# 统一设置默认 headers（所有请求生效）
s.headers.update({
    "Content-Type": "application/json",
    "User-Agent": "ApiTest/1.0",
})

# 统一设置默认参数（会与单次请求的 params 合并）
s.params.update({"appId": "1001"})

# 统一认证
s.auth = ("user", "pass")

# 统一超时（需通过适配器或封装，见 5.3）

# with 语句自动关闭
with requests.Session() as s:
    s.headers.update({"Authorization": f"Bearer {token}"})
    r = s.get(url)

5.3 用 Session 统一管理超时（封装技巧）

class TimeoutSession(requests.Session):
    """为所有请求设置默认超时"""
    def __init__(self, timeout=(3, 10)):
        super().__init__()
        self.timeout = timeout

    def request(self, *args, **kwargs):
        kwargs.setdefault("timeout", self.timeout)
        return super().request(*args, **kwargs)

s = TimeoutSession()
r = s.get(url)      # 自动带 timeout=(3, 10)

💡 性能提示：Session 的连接池复用在批量请求时能显著降低耗时（省去重复握手），压测/批量造数场景下尤为明显。

六、高级用法

6.1 重试机制

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

retry = Retry(
    total=3,                          # 总重试次数
    connect=3,                        # 连接失败重试次数
    read=2,                           # 读取失败重试次数
    status=3,                         # 状态码失败重试次数
    backoff_factor=0.5,               # 退避系数：0.5s → 1s → 2s
    status_forcelist=[429, 500, 502, 503, 504],   # 这些状态码触发重试
    allowed_methods=["GET", "POST"],  # 允许重试的方法（urllib3 2.x）
)

s = requests.Session()
s.mount("http://", HTTPAdapter(max_retries=retry))
s.mount("https://", HTTPAdapter(max_retries=retry))

r = s.get("https://api.example.com/flaky")

⚠️ 只对幂等请求（GET）或明确可重试的接口开启 POST 重试，避免重复下单。

6.2 连接池调优

adapter = HTTPAdapter(
    pool_connections=10,    # 连接池数量
    pool_maxsize=100,       # 单池最大连接
    max_retries=0,
)
s = requests.Session()
s.mount("https://", adapter)

6.3 钩子 Hooks

def log_response(r, *args, **kwargs):
    print(f"[{r.status_code}] {r.url} 耗时 {r.elapsed.total_seconds():.3f}s")

def assert_status(r, *args, **kwargs):
    r.raise_for_status()

s = requests.Session()
s.hooks["response"] = [log_response, assert_status]
s.get(url)    # 每个响应都会执行这两个回调

6.4 预处理请求 PreparedRequest

from requests import Request, Session

req = Request("POST", url, json={"a": 1}, headers={"X-Token": "abc"})
prepared = req.prepare()

print(prepared.method, prepared.url)
print(prepared.body)      # 可查看/修改最终发出的内容

s = Session()
r = s.send(prepared)

用途：调试”最终发出去的到底是什么”、构造带自定义编码的请求。

6.5 并发请求（批量造数/压测）

from concurrent.futures import ThreadPoolExecutor
import requests

urls = [f"https://api.example.com/user/{i}" for i in range(1, 101)]

def fetch(u):
    with requests.Session() as s:
        try:
            r = s.get(u, timeout=5)
            return r.status_code
        except requests.RequestException as e:
            return f"ERR: {e}"

with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(fetch, urls))

print(results.count(200))

requests 是同步库，并发需自行用线程池；若要原生异步可了解 httpx 或 aiohttp。

七、异常与错误处理

7.1 异常体系

异常	触发场景
requests.exceptions.RequestException	所有异常的基类
ConnectionError	连接失败（DNS、拒绝连接、网络不通）
Timeout	请求超时（细分 ConnectTimeout / ReadTimeout）
HTTPError	raise_for_status() 时状态码非 2xx
SSLError	SSL 证书校验失败
ProxyError	代理连接失败
TooManyRedirects	重定向次数超限
InvalidURL / MissingSchema	URL 格式错误（缺少 http://）
JSONDecodeError	.json() 解析失败

7.2 标准写法

import requests
from requests.exceptions import RequestException, Timeout, HTTPError

try:
    r = requests.get(url, timeout=(3, 10))
    r.raise_for_status()
    data = r.json()
except Timeout:
    print("请求超时")
except HTTPError as e:
    print(f"HTTP 错误：{e}")
except RequestException as e:
    print(f"请求失败：{e}")
else:
    print("成功：", data)

💡 自动化测试建议：不要让异常导致用例崩溃后无信息。捕获后 pytest.fail(f"请求失败: {e}") 或记录日志，报告里才能看清失败原因。

八、接口自动化实战（重点）

8.1 封装一个 API 客户端

# api_client.py
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class ApiClient:
    """接口测试客户端封装：统一 host、鉴权、超时、日志、断言"""

    def __init__(self, base_url, token=None, timeout=(3, 10)):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

        # 重试
        retry = Retry(total=2, backoff_factor=0.3,
                      status_forcelist=[500, 502, 503, 504],
                      allowed_methods=["GET"])
        self.session.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=20))

    def request(self, method, path, **kwargs):
        url = f"{self.base_url}/{path.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)
        r = self.session.request(method, url, **kwargs)
        # 统一打印（可替换为 logging）
        print(f"[{method}] {url} -> {r.status_code} ({r.elapsed.total_seconds():.3f}s)")
        return r

    def get(self, path, **kwargs):    return self.request("GET", path, **kwargs)
    def post(self, path, **kwargs):   return self.request("POST", path, **kwargs)
    def put(self, path, **kwargs):    return self.request("PUT", path, **kwargs)
    def delete(self, path, **kwargs): return self.request("DELETE", path, **kwargs)

    def close(self):
        self.session.close()

8.2 配合 pytest 使用

# test_api.py
import pytest
from api_client import ApiClient

BASE = "https://api.example.com"


@pytest.fixture(scope="session")
def client():
    c = ApiClient(BASE)
    yield c
    c.close()


@pytest.fixture(scope="session")
def token(client):
    r = client.post("/login", json={"username": "test01", "password": "123456"})
    assert r.status_code == 200
    return r.json()["data"]["token"]


def test_login_success(client):
    r = client.post("/login", json={"username": "test01", "password": "123456"})
    assert r.status_code == 200
    body = r.json()
    assert body["code"] == 0
    assert "token" in body["data"]


def test_login_wrong_password(client):
    r = client.post("/login", json={"username": "test01", "password": "wrong"})
    assert r.status_code == 200
    assert r.json()["code"] == 1001      # 业务错误码


def test_get_profile(client, token):
    client.session.headers["Authorization"] = f"Bearer {token}"
    r = client.get("/profile")
    assert r.status_code == 200
    assert r.json()["data"]["username"] == "test01"

8.3 接口关联（token / 数据传递）

# 场景：下单接口需要用到登录 token 和查询到的商品 ID

def test_create_order(client):
    # 1. 登录取 token
    r = client.post("/login", json={"username": "test01", "password": "123456"})
    token = r.json()["data"]["token"]
    client.session.headers["Authorization"] = f"Bearer {token}"

    # 2. 查商品取 ID
    r = client.get("/products", params={"keyword": "手机"})
    product_id = r.json()["data"]["list"][0]["id"]

    # 3. 下单（用上一步的数据）
    r = client.post("/orders", json={"productId": product_id, "qty": 1})
    assert r.status_code == 200
    order_id = r.json()["data"]["orderId"]

    # 4. 验证落库/查询
    r = client.get(f"/orders/{order_id}")
    assert r.json()["data"]["status"] == "CREATED"

8.4 数据驱动

import pytest

cases = [
    # (username, password, expected_code, desc)
    ("test01", "123456", 0,    "正常登录"),
    ("test01", "wrong",  1001, "密码错误"),
    ("",       "123456", 1002, "用户名为空"),
    ("test01", "",       1003, "密码为空"),
]

@pytest.mark.parametrize("username,password,expected_code,desc", cases)
def test_login_ddt(client, username, password, expected_code, desc):
    r = client.post("/login", json={"username": username, "password": password})
    assert r.json()["code"] == expected_code, f"{desc} 失败：{r.text}"

8.5 断言封装（统一校验）

def assert_api(resp, status=200, code=0, msg=None):
    """统一断言：状态码 + 业务码 + 可选消息"""
    assert resp.status_code == status, \
        f"状态码不符：预期 {status}，实际 {resp.status_code}，响应 {resp.text[:200]}"
    body = resp.json()
    assert body.get("code") == code, \
        f"业务码不符：预期 {code}，实际 {body.get('code')}，响应 {resp.text[:200]}"
    if msg is not None:
        assert msg in body.get("message", "")


# 使用
r = client.post("/login", json={...})
assert_api(r, status=200, code=0)

8.6 响应时间断言（性能基线）

def test_api_performance(client):
    r = client.get("/products", params={"page": 1})
    assert_api(r)
    # 单接口响应不超过 1 秒（冒烟性能基线）
    assert r.elapsed.total_seconds() < 1.0, f"响应过慢：{r.elapsed.total_seconds()}s"

九、签名 / 加密接口（测试常见难点）

9.1 常见签名流程

1. 把所有业务参数按 key 的字典序排序
2. 拼接成 "key1value1key2value2..." 形式
3. 首尾拼上 appSecret（或时间戳、nonce）
4. MD5 / SHA256 加密，转大写 → sign
5. 把 sign 和 timestamp、nonce 一起放进请求

9.2 MD5 签名实现

import hashlib
import time
import requests


def gen_sign(params: dict, secret: str) -> str:
    """MD5 签名：参数按 key 排序拼接 + secret"""
    sorted_items = sorted(params.items())
    raw = "".join(f"{k}{v}" for k, v in sorted_items)
    raw = secret + raw + secret
    return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()


def call_signed_api(url, app_key, app_secret, biz_params):
    params = {
        **biz_params,
        "appKey": app_key,
        "timestamp": str(int(time.time() * 1000)),
        "nonce": "abc123",
    }
    params["sign"] = gen_sign(params, app_secret)
    return requests.post(url, json=params, timeout=(3, 10))

9.3 其他常见加密

import hashlib
import hmac
import base64

# SHA256
hashlib.sha256("abc".encode()).hexdigest()

# HMAC-SHA256（常见于云厂商 API）
def hmac_sha256(secret: str, message: str) -> str:
    return hmac.new(secret.encode(), message.encode(), hashlib.sha256).hexdigest()

# Base64（图片、加密串传输）
base64.b64encode(b"hello").decode()      # aGVsbG8=
base64.b64decode("aGVsbG8=").decode()    # hello

# AES / RSA：需安装 pycryptodome
# pip install pycryptodome

💡 测试要点：签名接口最容易出的 Bug 是时间戳过期（客户端时间不准）、参数排序不一致（空值是否参与签名）、URL 编码差异。抓包对比自己生成的 sign 与 App 的 sign 是最快定位方式。

十、最佳实践与常见坑

坑	表现	解决
未设 timeout	脚本永久卡死	所有请求必传 timeout
data= / json= 用错	400/415	对照接口文档与 Content-Type
每次请求新建连接	批量任务很慢	用 Session 复用
中文乱码	text 显示乱码	r.encoding = r.apparent_encoding
verify=False 警告刷屏	控制台大量 Warning	urllib3.disable_warnings()
未关闭文件句柄	上传后文件被占用	用 with open()
断言只看状态码	业务错误码漏测	同时断言 code/message
异常吞掉	失败原因不明	捕获 RequestException 并打印
用例间状态污染	执行顺序影响结果	每个用例独立造数/清理

推荐实践：

1. 统一封装：base_url、鉴权、超时、日志集中管理（见 8.1）。
2. 配置外置：环境地址、账号放配置文件（配合 pytest.ini / .env），不要硬编码。
3. 用例独立：每个用例自己准备数据和清理，避免相互依赖。
4. 保留证据：失败时打印 r.request.url、r.request.body、r.text。
5. 断言分层：状态码 → 业务码 → 关键字段 → 数据落库校验。

十一、学习资源

资源	说明
requests 官方文档	权威，含快速上手与高级用法
requests 中文文档	中文翻译版
HTTPBin	在线 HTTP 测试服务，练手必备
JSONPlaceholder	免费假数据 REST API
curlconverter	cURL 转 requests 代码
《Python 自动化测试实战》	框架搭建向书籍