Python数据API超时处理怎么实现

wen python案例 2026-07-05 88

本文目录导读：

Python数据API超时处理怎么实现

requests库的超时处理
asyncio异步超时处理
使用装饰器处理超时
使用concurrent.futures实现超时
完整的超时处理示例
最佳实践建议

在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请求具有良好的超时处理机制,提高应用的稳定性和用户体验。

本文地址： https://www.bmjup.com/post/16382.html