代码解析
编程语言: Python
代码片段:
import requests
def fetch_json(url, retries=2, timeout=3):
for _ in range(retries + 1):
r = requests.get(url, timeout=timeout)
if r.ok:
return r.json()
return {}
功能与逻辑解析:
- 导入 requests 库用于发起 HTTP 请求。
- 定义函数 fetch_json(url, retries=2, timeout=3)：
- retries 表示重试次数，实际尝试次数为 retries+1（首次+重试）。
- 每次循环内调用 requests.get(url, timeout=timeout) 发起 GET 请求。
- 若响应 r.ok 为 True（HTTP 状态码 < 400，包括 2xx 和 3xx），立即返回解析的 JSON 内容 r.json()。
- 如果所有尝试都未满足 r.ok，则返回空字典 {}。
- 隐含行为与影响：
- r.ok 对 3xx（重定向）也为 True，但在自动重定向后若返回非 JSON 或空响应，会在 r.json() 处抛错。
- 未对网络异常进行捕获（如超时、连接错误），一旦发生异常会直接抛出并中断函数，循环重试不会生效。
- 未处理 JSON 解析异常（非 JSON 响应或空体会引发 ValueError/JSONDecodeError）。
- 对所有非 ok 状态码（含 404/400）都会盲目重试，可能造成不必要的请求。
调试建议或优化点:
- 增加异常处理与重试逻辑：
- 在 requests.get 外层捕获 requests.exceptions.RequestException，使网络错误（超时、连接重置、DNS 失败等）进行重试。
- 为 JSON 解析添加 try/except，避免非 JSON 响应导致崩溃。
- 改进重试策略：
- 使用指数退避 + 抖动（sleep）减少瞬时故障影响与拥塞。
- 仅对可恢复的状态码重试（如 429、500、502、503、504）；对 404/400 等客户端错误不重试。
- 明确响应校验：
- 使用 200 <= status_code < 300 严格限定成功范围，或在 ok 之后再检查 Content-Type 是否为 application/json 且内容非空。
- 明确返回类型与错误处理策略：
- 避免返回 {} 混淆“失败”与“服务端真实返回空对象”的情况。建议：
- 返回 None 表示失败，或
- 在耗尽重试后抛出异常，或
- 返回 (data, error) 元组。
- 补充类型注解与文档字符串，便于调用方处理。
- 连接复用与专业重试：
- 频繁调用建议使用 requests.Session() 复用连接。
- 需要更强重试能力可用 urllib3.util.retry.Retry + HTTPAdapter（对特定方法与状态码自动重试）。
- 超时更细化：
- 使用超时元组 timeout=(connect_timeout, read_timeout)，例如 (3.05, 10)，提升可控性。
- 其他：
实际用法示例:
# 原函数的直接使用与保护性调用
import requests
def fetch_json(url, retries=2, timeout=3):
for _ in range(retries + 1):
r = requests.get(url, timeout=timeout)
if r.ok:
return r.json()
return {}
# 示例 1：基础使用（注意：原函数未捕获异常，最好在外层 try/except）
try:
data = fetch_json("https://api.github.com")
if data:
print("成功获取：", list(data)[:5])
else:
print("请求未成功（返回空字典）")
except (requests.exceptions.RequestException, ValueError) as e:
print("请求或解析失败：", e)
# 示例 2：改进版（更稳健的重试与解析）
import time, random
from typing import Optional, Any
def fetch_json_safe(url: str, retries: int = 2, timeout=(3.05, 10)) -> Optional[Any]:
last_err = None
for attempt in range(1, retries + 2):
try:
r = requests.get(url, timeout=timeout)
# 严格成功判断
if 200 <= r.status_code < 300:
ctype = r.headers.get("Content-Type", "").lower()
if "application/json" in ctype and r.content:
return r.json()
return None # 成功但非 JSON 或空内容
# 仅对可恢复错误进行重试
if r.status_code in (429, 500, 502, 503, 504):
last_err = requests.HTTPError(f"HTTP {r.status_code}")
else:
r.raise_for_status()
except (requests.exceptions.RequestException, ValueError) as e:
last_err = e
# 剩余尝试则退避
if attempt <= retries:
sleep = min(2 ** (attempt - 1), 8) + random.random()
time.sleep(sleep)
if last_err:
# 根据需求：可改为 return None 或记录日志
raise last_err
# 使用改进版
try:
data = fetch_json_safe("https://api.github.com/rate_limit", retries=3, timeout=(3.05, 10))
print("改进版获取：", data is not None)
except Exception as e:
print("改进版仍失败：", e)
# 示例 3：使用 Session 与 Retry（更专业的 HTTP 层重试）
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def build_session_with_retry(total=3, backoff=0.5):
s = requests.Session()
retry = Retry(
total=total,
connect=total,
read=total,
status=total,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=("GET", "HEAD", "OPTIONS"),
backoff_factor=backoff,
raise_on_status=False,
)
s.mount("http://", HTTPAdapter(max_retries=retry))
s.mount("https://", HTTPAdapter(max_retries=retry))
return s
session = build_session_with_retry()
resp = session.get("https://api.github.com")
if 200 <= resp.status_code < 300:
print("Session 获取 JSON 长度：", len(resp.json()))
