本文目录导读：
在Python中处理数据API超时主要有以下几种方法：
requests库的超时处理
基础超时设置
import requests
from requests.exceptions import Timeout, ConnectionError
try:
# 设置连接超时和读取超时
response = requests.get('https://api.example.com/data',
timeout=(3, 5)) # (连接超时, 读取超时)
response.raise_for_status()
except Timeout:
print("请求超时")
except ConnectionError:
print("连接错误")
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
重试机制
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_retry_session():
session = requests.Session()
retry_strategy = Retry(
total=3, # 总重试次数
backoff_factor=1, # 退避因子（重试间隔会逐渐增加）
status_forcelist=[500, 502, 503, 504], # 需要重试的状态码
allowed_methods=["GET", "POST"] # 允许重试的方法
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
# 使用重试会话
session = create_retry_session()
try:
response = session.get('https://api.example.com/data', timeout=5)
response.raise_for_status()
except Exception as e:
print(f"最终失败: {e}")
asyncio异步超时处理
import asyncio
import aiohttp
from aiohttp import ClientTimeout
async def fetch_with_timeout():
timeout = ClientTimeout(total=10) # 总超时10秒
async with aiohttp.ClientSession(timeout=timeout) as session:
try:
async with session.get('https://api.example.com/data') as response:
return await response.json()
except asyncio.TimeoutError:
print("异步请求超时")
return None
except aiohttp.ClientError as e:
print(f"客户端错误: {e}")
return None
# 使用asyncio.run()运行
result = asyncio.run(fetch_with_timeout())
使用装饰器处理超时
import functools
import signal
from contextlib import contextmanager
class TimeoutError(Exception):
pass
@contextmanager
def timeout(seconds):
"""超时上下文管理器"""
def timeout_handler(signum, frame):
raise TimeoutError(f"操作超时（{seconds}秒）")
# 设置信号处理器
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0) # 取消定时器
def with_timeout(seconds):
"""超时装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
with timeout(seconds):
return func(*args, **kwargs)
except TimeoutError as e:
print(f"函数 {func.__name__} 超时: {e}")
return None
return wrapper
return decorator
# 使用示例
@with_timeout(5)
def fetch_api():
response = requests.get('https://api.example.com/data', timeout=3)
return response.json()
使用concurrent.futures实现超时
from concurrent.futures import ThreadPoolExecutor, TimeoutError
import requests
def fetch_data():
response = requests.get('https://api.example.com/data')
return response.json()
def fetch_with_concurrent_timeout(timeout_seconds=5):
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(fetch_data)
try:
result = future.result(timeout=timeout_seconds)
return result
except TimeoutError:
print("任务超时，已取消")
future.cancel() # 取消任务
return None
except Exception as e:
print(f"任务出错: {e}")
return None
完整的超时处理示例
import requests
import time
from functools import wraps
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class APIClient:
def __init__(self, base_url, default_timeout=3, max_retries=3):
self.base_url = base_url
self.default_timeout = default_timeout
self.max_retries = max_retries
self.session = self._create_session()
def _create_session(self):
"""创建带重试机制的会话"""
session = requests.Session()
retry_strategy = requests.packages.urllib3.util.retry.Retry(
total=self.max_retries,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
def get(self, endpoint, params=None, timeout=None):
"""带超时处理的GET请求"""
url = f"{self.base_url}/{endpoint}" if endpoint else self.base_url
timeout = timeout or self.default_timeout
start_time = time.time()
try:
logger.info(f"请求开始: {url}")
response = self.session.get(
url,
params=params,
timeout=(timeout // 2, timeout) # 连接超时为一半，读取超时为全部
)
response.raise_for_status()
elapsed = time.time() - start_time
logger.info(f"请求成功: {url} (耗时: {elapsed:.2f}s)")
return response.json()
except requests.exceptions.Timeout:
logger.error(f"请求超时: {url} (超时设置: {timeout}s)")
return {"error": "timeout", "message": f"请求在{timeout}秒内未完成"}
except requests.exceptions.RequestException as e:
logger.error(f"请求失败: {url} - {str(e)}")
return {"error": "request_failed", "message": str(e)}
def close(self):
"""关闭会话"""
self.session.close()
# 使用示例
def main():
client = APIClient(
base_url="https://api.example.com",
default_timeout=5,
max_retries=2
)
# 正常请求
result = client.get("data")
# 自定义超时
result = client.get("data", timeout=2)
client.close()
if __name__ == "__main__":
main()
最佳实践建议
设置合理的超时时间
- 连接超时: 通常1-3秒
- 读取超时: 根据API响应时间设置，通常5-30秒
- 总超时: 根据业务需求设置
超时策略选择
# 根据API类型设置不同超时
API_TIMEOUTS = {
"critical": (2, 5), # 关键API，短超时
"normal": (3, 10), # 普通API
"batch": (5, 30), # 批量API，长超时
"file_upload": (10, 60) # 文件上传，更长超时
}
监控和日志
import time
from contextlib import contextmanager
@contextmanager
def monitored_request(api_name, timeout):
"""监控请求的上下文管理器"""
start_time = time.time()
try:
yield
elapsed = time.time() - start_time
logger.info(f"API {api_name} 完成，耗时: {elapsed:.2f}s")
except TimeoutError:
elapsed = time.time() - start_time
logger.warning(f"API {api_name} 超时，耗时: {elapsed:.2f}s")
raise
异常处理层级
def fetch_data_with_retries(url, max_retries=3, timeout=5):
"""带重试的API请求"""
for attempt in range(max_retries):
try:
response = requests.get(url, timeout=timeout)
response.raise_for_status()
return response.json()
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError) as e:
wait_time = 2 ** attempt # 指数退避
logger.warning(f"第{attempt+1}次尝试失败: {e}")
if attempt < max_retries - 1:
time.sleep(wait_time)
else:
logger.error(f"所有重试均失败")
raise
这样实现可以确保你的API请求具有良好的超时处理机制,提高应用的稳定性和用户体验。
