你的API请求还在写裸套接字？requests焊死「Python网络请求」，从GET到Session一篇打通

你的 API 请求还在写裸套接字？requests 焊死「Python 网络请求」，从 GET 到 Session 一篇打通

痛点开场

上周我接手一个内部系统的数据同步脚本，打开代码一看差点晕过去：

import urllib.request
req = urllib.request.Request(
'https://api.internal.com/v1/users',
headers={'Authorization': 'Bearer ' + token}
)
response = urllib.request.urlopen(req, context=ssl.create_default_context())
data = json.loads(response.read().decode('utf-8'))

光是这 5 行代码，就藏着一堆麻烦：每个请求都要手写「建 Request → 传 context → read() → decode() → json.loads」五步样板，Cookie 不会自动保持，URL 参数得自己 urllib.parse.urlencode 拼，连超时都要另外传参。更恐怖的是后面还有 30 多个接口，每个都复制粘贴这套「裸 urllib」代码。

这还不是最惨的。我见过有人在生产环境里手写 socket 做 HTTP，处理 chunked encoding 写到凌晨三点；见过有人把用户名密码直接拼在 URL 里导致泄露；见过有人因为没设超时，一个挂掉的接口把整个批处理任务拖了 8 小时。

如果你还在用标准库的 urllib 做网络请求，或者在各种底层细节里反复踩坑——这篇就是给你写的。requests 不是「又一个 HTTP 库」，它是 Python 网络请求的工业标准。读完这篇，你会理解为什么有人说「HTTP for Humans」不是 slogan，而是事实。

第一章：requests 是什么？一句话说清楚

requests = Python 里发 HTTP 请求最像人话的方式。

它把 urllib、urllib2、httplib 那套「组装 Request 对象→处理上下文→解析响应」的流程，压缩成一行代码：

import requests
r = requests.get('https://httpbin.org/get')
print(r.status_code) # 200
print(r.json()) # 直接拿 JSON，不用 decode

核心概念就三个，理解了就能上手 90% 的场景：

ASCII 架构图：requests 在你代码里的位置

┌─────────────────────────────────────────────┐
│ 你的业务代码 │
│ (爬虫 / 对接 API / 测试脚本) │
└──────────────────┬────────────────────────────┘
│
┌─────────▼──────────┐
│ requests 库 │ ← 这篇的主角
│ (HTTP for Humans) │
└─────────┬──────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ urllib3 │ │ certifi │ │ charset │
│ (连接池) │ │ (证书库) │ │ (编码) │
└──────────┘ └──────────┘ └──────────┘
│
┌─────────▼──────────┐
│ TCP / HTTP 协议层 │
└─────────────────────┘

urllib3 负责真正的网络连接和连接池复用，certifi 提供 CA 证书 bundle，charset_normalizer 处理编码探测。requests 自己不干重活，它做的是「把底层能力封装成你能一眼看懂的 API」。

第二章：基础请求——GET/POST/PUT/DELETE 完整代码

2.1 四种 HTTP 方法的完整示例

import requests

base_url = "https://httpbin.org"

# ── GET：获取数据 ──
response = requests.get(
f"{base_url}/get",
params={"page": 1, "limit": 10},
timeout=10
)
print(f"GET 状态码: {response.status_code}")
print(f"GET URL: {response.url}") # 自动拼接参数
print(f"GET 响应: {response.json()}")

# ── POST：提交 JSON 数据 ──
payload = {"username": "test_user", "password": "secret123"}
response = requests.post(
f"{base_url}/post",
json=payload, # 自动加 Content-Type: application/json
timeout=10
)
print(f"POST 状态码: {response.status_code}")
print(f"POST 响应: {response.json()}")

# ── PUT：更新资源 ──
update_data = {"name": "new_name", "status": "active"}
response = requests.put(
f"{base_url}/put",
json=update_data,
timeout=10
)
print(f"PUT 状态码: {response.status_code}")

# ── DELETE：删除资源 ──
response = requests.delete(
f"{base_url}/delete",
timeout=10
)
print(f"DELETE 状态码: {response.status_code}")

2.2 进阶：表单提交、文件上传、自定义 Header

import requests

# ── 表单提交 (application/x-www-form-urlencoded) ──
response = requests.post(
"https://httpbin.org/post",
data={"username": "admin", "password": "123456"},
timeout=10
)

# ── 自定义 Headers + 代理 ──
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
"Accept": "application/json",
"X-Custom-Token": "my-secret-token"
}

proxies = {
"http": "http://127.0.0.1:8080",
"https": "http://127.0.0.1:8080",
}

response = requests.get(
"https://httpbin.org/headers",
headers=headers,
proxies=proxies,
timeout=10
)

# ── 文件上传 ──
with open("report.pdf", "rb") as f:
response = requests.post(
"https://httpbin.org/post",
files={"file": ("report.pdf", f, "application/pdf")},
timeout=30
)
print(f"文件上传状态: {response.status_code}")

第三章：进阶功能——Session、认证、代理、超时、重试

3.1 Session 对象：保持连接复用和状态

import requests

# 创建 Session，自动处理 Cookie 和连接复用
session = requests.Session()

# 设置全局默认 headers
session.headers.update({
"User-Agent": "MyApp/1.0",
"Accept": "application/json"
})

# 先登录获取 Cookie
login_resp = session.post(
"https://httpbin.org/post",
json={"user": "admin", "pass": "secret"},
timeout=10
)

# 后续请求自动携带 Cookie，且复用底层 TCP 连接
profile_resp = session.get("https://httpbin.org/get", timeout=10)
orders_resp = session.get("https://httpbin.org/get", timeout=10)

# 性能对比：Session vs 独立请求
# Session：TCP 连接复用，3 个请求共用 1 条连接
# 独立 requests.get()：每个请求新建 TCP 连接 + SSL 握手

session.close()

3.2 认证方式：Basic Auth / Bearer Token / 自定义

import requests
from requests.auth import HTTPBasicAuth

# Basic Auth（用户名+密码）
response = requests.get(
"https://httpbin.org/basic-auth/admin/secret",
auth=HTTPBasicAuth("admin", "secret"),
timeout=10
)

# Bearer Token（现代 API 主流方式）
token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9…"
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(
"https://api.github.com/user",
headers=headers,
timeout=10
)

# 自定义认证（继承 AuthBase）
from requests.auth import AuthBase

class TokenAuth(AuthBase):
def __init__(self, token):
self.token = token
def __call__(self, r):
r.headers['X-API-Key'] = self.token
return r

response = requests.get(
"https://httpbin.org/headers",
auth=TokenAuth("my-api-key-123"),
timeout=10
)

3.3 超时、重试与错误处理（生产环境必备）

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

# 配置自动重试策略
retry_strategy = Retry(
total=3, # 最多重试 3 次
backoff_factor=1, # 退避间隔 1s、2s、4s 递增（尊重服务端 Retry-After 头）
status_forcelist=[429, 500, 502, 503, 504], # 遇到这些状态码才重试
# urllib3 2.x 用 allowed_methods（旧版叫 method_whitelist）；
# 默认只重试幂等方法，POST 默认不重试——把 POST 加进来的前提是
# 你的接口幂等（重试不会重复下单/重复扣款），否则别加
allowed_methods=["HEAD", "GET", "PUT", "DELETE"]
)

adapter = HTTPAdapter(max_retries=retry_strategy)

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

# 现在请求会自动处理重试
response = session.get(
"https://httpbin.org/status/503",
timeout=(3.05, 27), # (连接超时, 读取超时)
)

# ── 完整的错误处理模板 ──
from requests.exceptions import (
RequestException, Timeout, ConnectionError,
HTTPError, TooManyRedirects
)

def safe_request(url, method="get", **kwargs):
"""生产级安全请求模板"""
try:
resp = session.request(method, url, **kwargs)
resp.raise_for_status() # 4xx/5xx 自动抛 HTTPError
return resp
except Timeout:
print(f"[{url}] 请求超时")
return None
except ConnectionError:
print(f"[{url}] 连接失败")
return None
except HTTPError as e:
status = getattr(e.response, "status_code", "未知")
print(f"[{url}] HTTP 错误: {status}")
return None
except TooManyRedirects:
print(f"[{url}] 重定向过多")
return None
except RequestException as e:
print(f"[{url}] 请求异常: {e}")
return None

# 使用
result = safe_request("https://api.example.com/data", "get", timeout=10)
if result:
print(result.json())

第四章：三个进阶场景

4.1 场景一：批量接口测试脚本

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

# 待测试的接口列表
endpoints = [
{"method": "GET", "url": "https://httpbin.org/get"},
{"method": "POST", "url": "https://httpbin.org/post", "json": {"test": 1}},
{"method": "PUT", "url": "https://httpbin.org/put", "json": {"update": True}},
{"method": "DELETE", "url": "https://httpbin.org/delete"},
]

def test_endpoint(item):
"""测试单个接口并返回结果"""
# 在 try 外先取出标识字段，避免取值就抛异常时，except 里引用未定义变量
method = item.get("method", "GET").lower()
url = item.get("url", "<unknown>")
try:
kwargs = {k: v for k, v in item.items() if k not in ["method", "url"]}
kwargs["timeout"] = 10

# 注意：模块级 requests.request 每次都会新建并销毁一个临时 Session，
# 线程池里无法复用连接。生产写法是每个线程持有自己的 Session
# （threading.local 缓存），或直接用后面的 requests-futures
resp = requests.request(method, url, **kwargs)
return {
"url": url,
"method": item["method"],
"status": resp.status_code,
"elapsed_ms": resp.elapsed.total_seconds() * 1000,
"ok": resp.ok
}
except Exception as e:
return {"url": url, "method": item["method"], "error": str(e), "ok": False}

# 并发测试
with ThreadPoolExecutor(max_workers=4) as executor:
futures = [executor.submit(test_endpoint, ep) for ep in endpoints]

print("=" * 60)
print("批量接口测试结果")
print("=" * 60)

for future in as_completed(futures):
result = future.result()
status = "✓" if result.get("ok") else "✗"
if "error" in result:
print(f"{status} [{result['method']}] {result['url']} – 错误: {result['error']}")
else:
print(f"{status} [{result['method']}] {result['status']} {result['elapsed_ms']:.0f}ms {result['url']}")

4.2 场景二：流式下载大文件

import requests
from pathlib import Path

def download_large_file(url: str, save_path: str, chunk_size: int = 8192):
"""流式下载大文件，避免内存爆炸"""
save_path = Path(save_path)
save_path.parent.mkdir(parents=True, exist_ok=True)

with requests.get(url, stream=True, timeout=30) as resp:
resp.raise_for_status()
total_size = int(resp.headers.get("content-length", 0))
downloaded = 0

with open(save_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size:
percent = (downloaded / total_size) * 100
print(f"\\r下载进度: {percent:.1f}%", end="", flush=True)

print(f"\\n✓ 下载完成: {save_path} ({downloaded} bytes)")
return save_path

# 使用示例
# download_large_file(
# "https://example.com/large-dataset.zip",
# "./downloads/dataset.zip"
# )

4.3 场景三：异步请求与 requests 生态（aiohttp / requests-futures）

# 方案 A：requests-futures（线程池包装，最简单）
from requests_futures.sessions import FuturesSession

session = FuturesSession(max_workers=10)

# 提交异步请求，立即返回 future
future1 = session.get("https://httpbin.org/delay/1")
future2 = session.get("https://httpbin.org/delay/1")
future3 = session.get("https://httpbin.org/delay/1")

# 按需获取结果（阻塞等待）
resp1 = future1.result()
resp2 = future2.result()
resp3 = future3.result()

print("三个 1 秒延迟的请求并发执行，总耗时约 1 秒，而不是串行的 3 秒")

# 方案 B：aiohttp（真正的异步，适合高并发）
import aiohttp
import asyncio

async def fetch_one(session, url):
# 关键：session.get() 返回异步上下文管理器，必须用 async with 接住，
# 并在上下文内把响应体读完；直接 await session.get(url) 拿到 response
# 却不退出上下文，连接不会归还连接池，高并发下很快把连接池耗尽
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return await resp.json() # 按实际需要选 text() / read() / json()

async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_one(session, url) for url in urls]
return await asyncio.gather(*tasks)

# asyncio.run(fetch_all(["https://httpbin.org/get"] * 10))

第五章：8 个暗坑——requests 排雷表

特别说明：关于 verify=False 的安全红线

我见过太多教程和 StackOverflow 答案把 requests.get(url, verify=False) 当成「解决 SSL 报错」的灵丹妙药。这不是解法，是埋雷。注意一个常见误解：verify=False 不会让流量变成明文，TLS 加密通道仍然建立；它真正关掉的是「证书身份校验」——客户端不再确认对端证书是不是由受信 CA 签给目标域名的，于是中间人用自签证书冒充服务器时客户端毫无察觉，攻击者由此可以解密、篡改你的流量。正确做法：自签证书就把 CA 证书路径传给 verify='/path/to/ca.crt'，企业内网就统一分发根证书；生产环境保持 verify=True。临时用 verify=False 排障时可以配合 urllib3.disable_warnings() 消噪，但这段代码绝不能进生产。

第六章：8 道面试 Q&A

Q1：requests 和 urllib 的核心区别是什么？

requests 是「HTTP for Humans」——API 设计符合直觉，自动处理编码、Cookie、连接复用。urllib 是标准库，功能完整但 API 繁琐，需要手动拼装 Request 对象、处理 SSL 上下文、解码响应。

Q2：Session 对象解决了什么问题？

三个问题：① 连接复用（TCP + SSL 握手只做一次）② Cookie 持久化（自动在请求间携带）③ 配置复用（headers、auth、proxies 只需设一次）。性能上 Session 比独立请求快数倍。

Q3：GET 和 POST 在 requests 里传参有什么区别？

GET 用 params=，数据拼在 URL 问号后；POST 用 data=（表单编码）或 json=（JSON 编码），数据在请求体里。混用会导致服务端收不到参数。

Q4：如何处理请求超时？

timeout=(3.05, 27) —— 第一个是连接超时（TCP 握手），第二个是读取超时（等响应体）。永远设超时，否则网络抖动会让程序无限 hang。

Q5：怎么实现自动重试？

用 urllib3.util.retry.Retry + requests.adapters.HTTPAdapter，配置重试次数、退避策略、触发状态码，然后通过 session.mount() 生效。比手写 try/except 循环可靠得多。

Q6：下载 1GB 文件怎样避免内存溢出？

stream=True + iter_content(chunk_size=8192) 分块读取，写磁盘而不是进内存。配合 with 语句确保连接关闭。

Q7：requests 是线程安全的吗？

requests 模块本身是线程安全的，但 Session 不是完全线程安全——多个线程共用一个 Session 时，Cookie 更新可能冲突。高并发场景建议每个线程一个 Session，或直接用线程池的 FuturesSession。

Q8：requests 和 aiohttp 怎么选？

IO 密集型 + 高并发（成百上千路并发连接）→ aiohttp（协程异步，单线程即可维持大量等待中的连接，线程开销远小于线程池方案）。普通场景、快速开发、已有同步代码 → requests（生态成熟，调试简单）。两者可以混用：核心逻辑用 requests，确认是并发瓶颈的环节再换 aiohttp。

第七章：总结 + 判断标准 + 参考资料

核心知识点回顾

3 条万能判断标准

什么情况不该用 requests

如果你的项目已经是全异步架构（FastAPI + asyncio），到处混用 requests（同步阻塞）会拖垮整个事件循环。这时候该用 httpx（兼容 requests API 的异步库）或 aiohttp。另外，需要 HTTP/2 协议支持时，requests 本身不支持，需要换 httpx。

参考资料

• 官方文档：Requests: HTTP for Humans —— 最权威，示例丰富
• urllib3 文档：urllib3.readthedocs.io —— 想了解连接池和重试机制必读
• 进阶替代：httpx 库（兼容 requests API + 原生异步 + HTTP/2 支持）
• 真实项目参考：GitHub 上搜索 “requests session best practices” 有大量生产级封装示例

封面动物：壁虎 —— 寓意"敏捷攀爬各种协议与接口的灵活网络爬虫，像 requests 库灵活应对各类 HTTP 请求"

模型能力和 SDK API 更新很快，本文代码基于当前主流实现，实际调用时请核对最新版本接口。