最佳实践
下面是一份 Python Requests 最佳实践(专业级),包含工程化、稳定性、可维护性、安全性、性能全方面的总结。
❌ 不要每次都 requests.get
requests.get(url)
requests.post(url)
✅ 用 Session() 保持 TCP 连接 + 自动复用 Cookie + 提高性能
import requests
session = requests.Session()
session.headers.update({"User-Agent": "MyApp/1.0"})
def get(url, **kwargs):
return session.get(url, timeout=(3, 10), **kwargs)
r = get("https://example.com")
Session 能减少 80% 以上的 TCP 建立开销,适合高频 API 调用和爬虫。
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import requests
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504],
allowed_methods=["GET", "POST"],
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)
📌 好处
- 避免偶发网络故障
- 自动重试 HTTP 5xx
- backoff 防止请求风暴
❌ 这会卡住一整天
requests.get(url)
✅ 正确写法
requests.get(url, timeout=(3, 10))
解释:
- 3 -> 连接超时(TCP 连接)
- 10 -> 读取超时(服务器处理时间)
生产环境必须写!
🧱 最佳实践:初始化 Session 时统一设定
session = requests.Session()
session.headers.update({
"User-Agent": "MyApp/1.0",
"Authorization": f"Bearer {token}",
})
避免每个请求重复传。
❌ 错误方式(文件大于 1G 会炸内存)
r = requests.get(url)
open("a.zip", "wb").write(r.content)
✅ 正确方式
with requests.get(url, stream=True) as r:
r.raise_for_status()
with open("a.zip", "wb") as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
❌ 老写法
if r.status_code != 200:
print("请求失败")
✅ 新写法
try:
r = requests.get(url)
r.raise_for_status()
except requests.RequestException as e:
print("请求失败:", e)
一次性处理所有网络、状态码异常。
❌ 错误写法(发送 x-www-form-urlencoded)
requests.post(url, data={"a": 1})
✅ 正确写法(标准 JSON 请求)
requests.post(url, json={"a": 1})
requests 会自动加 Content-Type: application/json。
完整结构应该是:
import requests
try:
r = session.get(url, timeout=(3, 10))
r.raise_for_status()
data = r.json()
except requests.Timeout:
print("请求超时")
except requests.ConnectionError:
print("网络连接失败")
except requests.HTTPError as e:
print("HTTP 错误:", e)
except ValueError:
print("JSON 解析失败")
except Exception as e:
print("未知错误:", e)
高质量生产环境必备。
Charles / Fiddler / mitmproxy 调试必需:
proxies = {
"http": "http://127.0.0.1:8888",
"https": "http://127.0.0.1:8888",
}
session.get(url, proxies=proxies, verify=False)
r = session.get(url)
print("URL:", r.request.url)
print("Method:", r.request.method)
print("Request Headers:", r.request.headers)
print("Request Body:", r.request.body)
print("Status:", r.status_code)
print("Response Headers:", r.headers)
配合 API 调试非常有用。
import os
TOKEN = os.getenv("API_TOKEN")
避免 Token 泄漏。
建议在工程中封成类:
class APIClient:
def __init__(self, base_url: str, token: str = None):
self.session = requests.Session()
self.base_url = base_url
self.session.headers.update({
"User-Agent": "MyApp/1.0",
})
if token:
self.session.headers["Authorization"] = f"Bearer {token}"
# 加上重试
retry = Retry(total=3, backoff_factor=0.3, allowed_methods=["GET","POST"])
adapter = HTTPAdapter(max_retries=retry)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
def get(self, path, **kwargs):
return self.session.get(self.base_url + path, timeout=(3,10), **kwargs)
def post(self, path, **kwargs):
return self.session.post(self.base_url + path, timeout=(3,10), **kwargs)
之后所有接口变得非常简洁:
client = APIClient("https://api.example.com", token="jwt_token_here")
data = client.get("/user/info").json()
这是企业级项目标准写法。
requests.get(url, verify=False)
⚠️ 生产环境不要关闭 SSL!
配合 loguru:
logger.info(f"Request -> {url}, params={params}")
logger.info(f"Response <- {r.status_code}, body={r.text}")
API Debug 非常方便。
from concurrent.futures import ThreadPoolExecutor
import requests
session = requests.Session()
def fetch(url):
return session.get(url).text
with ThreadPoolExecutor(max_workers=10) as pool:
results = list(pool.map(fetch, urls))
高效又简单。
| 项目 | 最佳实践 |
|---|---|
| 1 | 用 Session(),不要每次 requests.get |
| 2 | 强制使用 timeout |
| 3 | 加重试机制(Retry) |
| 4 | 用 .raise_for_status() |
| 5 | JSON 请求用 json= |
| 6 | 流式下载大文件 |
| 7 | 优雅的异常捕获结构 |
| 8 | 所有 headers/Tokens 集中设置 |
| 9 | 用 Proxy 做抓包/调试 |
| 10 | 打印请求/响应用于调试 |
| 11 | 敏感信息用环境变量 |
| 12 | 工程级封装成 API Client |
示例:
client = APIClient(base_url="https://example.com/api")
client.session.headers.update({"User-Agent": "Mozilla/5.0 ..."})
response = client.get(f"?id=100")
if response is None:
exit()
response_dict: dict = response.json()
data: list[dict] = response_dict.get("data", [])