Python Requests库的最佳实践、性能优化和技巧
Python Requests库最佳实践与技巧摘要 本文总结了Python Requests库的核心使用技巧,涵盖安全、性能优化和高级功能三大方面: 基础安全实践 必须设置超时(timeout)防止阻塞 使用raise_for_status()检查响应状态码 生产环境务必验证SSL证书 性能优化 使用Session对象复用TCP连接 通过ThreadPoolExecutor实现并发请求 添加请求间
Python Requests库的最佳实践、性能优化和技巧
Python Requests 库虽然易用,但在实际开发中(尤其是高并发、复杂接口交互、爬虫等场景),掌握一些最佳实践和技巧能显著提升代码的可靠性、性能和可维护性。以下是经过实践验证的核心技巧:
一、基础安全与可靠性实践
1. 必须设置超时(Timeout)
问题:未设置超时的请求可能因网络问题无限阻塞,导致程序挂起。
解决:为所有请求设置超时时间(推荐拆分为「连接超时」和「读取超时」)。
import requests
# 最佳实践:分别设置连接超时(建立TCP连接的时间)和读取超时(获取响应数据的时间)
try:
# 连接超时5秒,读取超时10秒(根据业务调整)
response = requests.get(
"https://api.example.com/data",
timeout=(5, 10) # (connect_timeout, read_timeout)
)
except requests.exceptions.Timeout:
print("请求超时:连接或读取数据超时")
except requests.exceptions.RequestException as e:
print(f"其他请求错误:{e}")
2. 始终检查响应状态码
问题:默认情况下,即使返回 4xx(客户端错误)或 5xx(服务器错误),Requests 也不会主动抛出异常,可能导致错误数据被误用。
解决:使用 response.raise_for_status()
自动抛出 HTTP 错误,或显式检查状态码。
response = requests.get("https://api.example.com/data")
# 方式1:主动抛出HTTP错误(推荐)
try:
response.raise_for_status() # 状态码4xx/5xx时抛出HTTPError
except requests.exceptions.HTTPError as e:
print(f"HTTP错误:{e}")
# 方式2:显式检查状态码(适合需要自定义处理的场景)
if response.status_code == 200:
print("请求成功")
elif response.status_code == 404:
print("资源不存在")
else:
print(f"请求失败,状态码:{response.status_code}")
3. 验证SSL证书(生产环境必做)
问题:禁用 SSL 验证会导致安全风险(如中间人攻击)。
解决:默认启用 SSL 验证,仅在测试环境临时关闭(需明确指定 verify=False
)。
# 生产环境:默认验证SSL证书(推荐)
response = requests.get("https://api.example.com") # verify=True(默认)
# 测试环境:临时关闭验证(需谨慎,仅在信任的环境使用)
response = requests.get("https://internal-test-api", verify=False)
# 关闭验证时会有警告,可通过以下方式屏蔽(不推荐在生产环境)
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
进阶:若使用自签名证书,可指定证书路径:response = requests.get("https://api.example.com", verify="/path/to/cert.pem")
二、性能优化技巧
1. 使用 Session 复用连接
问题:频繁调用 requests.get()
会重复建立 TCP 连接,增加开销。
解决:使用 requests.Session()
复用连接(HTTP 持久连接),尤其适合多次访问同一域名的场景。
# 低效方式:每次请求新建连接
for _ in range(10):
requests.get("https://api.example.com/data") # 每次新建TCP连接
# 高效方式:复用连接
with requests.Session() as session: # 上下文管理器自动关闭会话
for _ in range(10):
session.get("https://api.example.com/data") # 复用同一TCP连接
补充:Session 还会自动维护 cookies,适合模拟登录状态(如先 POST 登录,再 GET 需认证的资源)。
2. 批量请求处理(并发优化)
问题:同步请求在批量操作时效率低下(如爬取多个页面)。
解决:结合线程池实现并发请求(无需切换到异步库)。
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
def fetch_url(url):
"""单个URL请求函数"""
try:
response = requests.get(url, timeout=(5, 10))
response.raise_for_status()
return {"url": url, "status": "success", "data": response.text[:100]}
except Exception as e:
return {"url": url, "status": "failed", "error": str(e)}
# 批量URL列表
urls = [f"https://httpbin.org/get?page={i}" for i in range(1, 21)]
# 并发请求(线程池大小根据目标服务器承受能力调整,建议5-20)
with ThreadPoolExecutor(max_workers=10) as executor:
# 提交所有任务
futures = {executor.submit(fetch_url, url): url for url in urls}
# 获取结果(按完成顺序)
for future in as_completed(futures):
result = future.result()
print(f"{result['url']}:{result['status']}")
3. 控制请求频率(避免过载或反爬)
问题:高频请求可能触发服务器反爬机制或导致服务过载。
解决:添加请求间隔(适合爬虫),或使用令牌桶算法控制速率。
import time
import requests
from itertools import cycle
# 简单间隔控制(适合低频场景)
urls = [f"https://httpbin.org/get?i={i}" for i in range(10)]
for url in urls:
response = requests.get(url)
print(f"请求 {url},状态码:{response.status_code}")
time.sleep(1) # 间隔1秒
# 进阶:使用代理池+随机间隔(反爬场景)
proxies_list = [
"http://proxy1:8080",
"http://proxy2:8080"
]
proxy_pool = cycle(proxies_list) # 循环使用代理
for url in urls:
proxy = next(proxy_pool)
try:
response = requests.get(
url,
proxies={"http": proxy, "https": proxy},
timeout=5
)
print(f"使用代理 {proxy} 请求成功")
except:
print(f"代理 {proxy} 失效,切换下一个")
time.sleep(0.5 + random.random()) # 随机间隔0.5-1.5秒
三、高级功能与场景技巧
1. 自定义请求适配器(处理重试逻辑)
问题:网络波动可能导致临时失败,需要自动重试。
解决:使用 urllib3
的 HTTPAdapter
结合 Retry
实现自动重试。
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 配置重试策略
retry_strategy = Retry(
total=3, # 总重试次数
backoff_factor=1, # 重试间隔:{backoff_factor} * (2 **({重试次数} - 1))
status_forcelist=[429, 500, 502, 503, 504], # 需要重试的状态码
allowed_methods=["GET", "POST"] # 允许重试的HTTP方法
)
# 创建适配器并挂载到Session
adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("https://", adapter) # 为HTTPS请求挂载适配器
session.mount("http://", adapter) # 为HTTP请求挂载适配器
# 使用带重试的Session请求
response = session.get("https://api.example.com/unstable-endpoint")
2.** 处理大型响应(流式传输)问题:直接获取大文件(如视频、压缩包)会占用大量内存。
解决:使用 stream=True
流式读取,分块写入本地。
# 下载大文件(流式传输)
url = "https://example.com/large_file.zip"
local_filename = "large_file.zip"
with requests.get(url, stream=True, timeout=30) as response:
response.raise_for_status()
# 分块写入(每次10MB)
with open(local_filename, "wb") as f:
for chunk in response.iter_content(chunk_size=10*1024*1024): # 10MB
if chunk: # 过滤空块
f.write(chunk)
print(f"文件 {local_filename} 下载完成")
3. 模拟浏览器行为(反爬与兼容性)
问题:服务器可能通过 User-Agent
、Referer
等头信息识别非浏览器请求并拒绝。
解决:设置真实的请求头,模拟浏览器行为。
# 模拟Chrome浏览器的请求头(从浏览器开发者工具复制真实头信息)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
"Referer": "https://www.google.com/", # 模拟从Google跳转
"Connection": "keep-alive"
}
response = requests.get("https://example.com", headers=headers)
进阶:随机切换 User-Agent
(使用 fake_useragent
库):
from fake_useragent import UserAgent
ua = UserAgent()
headers = {"User-Agent": ua.random} # 随机生成一个浏览器UA
4. 复杂认证处理(Basic、Digest、OAuth)
Requests 内置多种认证方式,无需手动构造 Authorization
头。
# 1. Basic认证(用户名+密码)
from requests.auth import HTTPBasicAuth
response = requests.get(
"https://api.example.com/protected",
auth=HTTPBasicAuth("username", "password")
)
# 2. Digest认证
from requests.auth import HTTPDigestAuth
response = requests.get(
"https://api.example.com/digest-protected",
auth=HTTPDigestAuth("username", "password")
)
# 3. OAuth1认证(需安装requests-oauthlib)
from requests_oauthlib import OAuth1
auth = OAuth1("client_key", "client_secret", "resource_owner_key", "resource_owner_secret")
response = requests.get("https://api.example.com/oauth1", auth=auth)
四、调试与日志技巧
1. 启用详细日志(调试请求过程)
通过配置日志,查看请求的详细信息(URL、头信息、响应等),方便排查问题。
import logging
import requests
# 配置日志级别为DEBUG,输出请求细节
logging.basicConfig(level=logging.DEBUG)
# 禁用urllib3的DEBUG日志(避免过于冗余)
logging.getLogger("urllib3").setLevel(logging.WARNING)
response = requests.get("https://httpbin.org/get")
输出示例:
DEBUG:requests.packages.urllib3.connectionpool:Starting new HTTPS connection (1): httpbin.org:443
DEBUG:requests.packages.urllib3.connectionpool:https://httpbin.org:443 "GET /get HTTP/1.1" 200 239
2. 使用 httpbin.org 测试请求
httpbin.org 是一个专门用于测试 HTTP 请求的服务,可返回请求的详细信息(头信息、参数、cookies 等)。
# 测试POST请求数据
response = requests.post(
"https://httpbin.org/post",
json={"name": "test"},
headers={"X-Test": "123"}
)
print(response.json()) # 查看服务器收到的请求数据,验证是否符合预期
五、常见陷阱与避坑指南
-
不要忽略
response.encoding
Requests 会自动猜测编码,但可能出错(尤其是非 UTF-8 页面)。可手动指定:response.encoding = "gbk"
(针对中文页面常见的 GBK 编码)。 -
避免在循环中创建 Session
每次创建 Session 会重新初始化连接池,抵消复用连接的优势。应在循环外创建 Session 并复用。 -
注意
json()
方法的异常
若响应不是 JSON 格式,response.json()
会抛出JSONDecodeError
,需捕获:try: data = response.json() except ValueError: print("响应不是JSON格式")
-
谨慎处理
allow_redirects
默认情况下 Requests 会自动跟随重定向(allow_redirects=True
)。若需禁止重定向(如获取重定向 URL),可设置allow_redirects=False
。
总结
Requests 库的最佳实践核心可概括为:安全第一(超时、SSL验证)、性能优化(Session复用、并发)、模拟真实(请求头、频率控制)、容错处理(重试、异常捕获)。掌握这些技巧能让你在接口开发、爬虫、数据采集等场景中写出更可靠、高效的代码。
根据具体场景灵活调整策略(如爬虫需侧重反爬技巧,企业接口需侧重稳定性和认证),才能充分发挥 Requests 的优势。
更多推荐
所有评论(0)