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. 自定义请求适配器(处理重试逻辑)

问题:网络波动可能导致临时失败,需要自动重试。
解决:使用 urllib3HTTPAdapter 结合 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-AgentReferer 等头信息识别非浏览器请求并拒绝。
解决:设置真实的请求头,模拟浏览器行为。

# 模拟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())  # 查看服务器收到的请求数据,验证是否符合预期

五、常见陷阱与避坑指南

  1. 不要忽略 response.encoding
    Requests 会自动猜测编码,但可能出错(尤其是非 UTF-8 页面)。可手动指定:
    response.encoding = "gbk"(针对中文页面常见的 GBK 编码)。

  2. 避免在循环中创建 Session
    每次创建 Session 会重新初始化连接池,抵消复用连接的优势。应在循环外创建 Session 并复用。

  3. 注意 json() 方法的异常
    若响应不是 JSON 格式,response.json() 会抛出 JSONDecodeError,需捕获:

    try:
        data = response.json()
    except ValueError:
        print("响应不是JSON格式")
    
  4. 谨慎处理 allow_redirects
    默认情况下 Requests 会自动跟随重定向(allow_redirects=True)。若需禁止重定向(如获取重定向 URL),可设置 allow_redirects=False

总结

Requests 库的最佳实践核心可概括为:安全第一(超时、SSL验证)、性能优化(Session复用、并发)、模拟真实(请求头、频率控制)、容错处理(重试、异常捕获)。掌握这些技巧能让你在接口开发、爬虫、数据采集等场景中写出更可靠、高效的代码。

根据具体场景灵活调整策略(如爬虫需侧重反爬技巧,企业接口需侧重稳定性和认证),才能充分发挥 Requests 的优势。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐