Python API 请求中的异常处理设计

答案：Python API请求异常处理需分层捕获连接、超时、HTTP错误及解析异常，结合指数退避重试机制，并通过日志记录与自定义异常提升可维护性。

在Python进行API请求时，异常处理设计绝非可有可无的“锦上添花”，它实际上是构建任何健壮、可靠系统的基石。说白了，网络环境复杂多变，远程服务也并非总是尽如人意，请求失败是常态而非意外。因此，我们的核心观点是：一套深思熟虑的异常处理机制，能显著提升应用的韧性、用户体验，并极大地简化后期调试与维护工作。 它不只是捕获错误，更是关于如何优雅地应对、恢复，甚至在无法恢复时，也能提供清晰的反馈。

设计Python API请求的异常处理，我个人通常会从几个层面去考量：识别潜在的失败点，选择合适的捕获策略，以及在错误发生后如何进行有效的恢复或通知。这其实是一个从“可能发生什么”到“发生后怎么办”的完整闭环。

Python API请求中常见的异常类型有哪些，以及如何针对性捕获？

在处理Python的API请求时，尤其是使用

requests

try...except Exception

最常见的，也是最基础的，莫过于网络连接问题。当你的程序无法连接到目标服务器时，比如DNS解析失败、网络不通，或者远程主机拒绝连接，

requests.exceptions.ConnectionError

立即学习“Python免费学习笔记（深入）”；

import requests

try:
    response = requests.get('http://nonexistent-domain.com')
    response.raise_for_status() # 这行代码会检查HTTP状态码，如果不是2xx，就抛出HTTPError
except requests.exceptions.ConnectionError as e:
    print(f"连接错误：无法连接到服务器。请检查网络或目标地址。错误详情: {e}")
    # 这里通常会记录日志，或者通知运维人员

紧接着，超时问题也是家常便饭。如果API响应时间过长，我们不想让用户一直傻等，或者耗尽服务器资源，就需要设置超时。一旦超过设定的时间，

requests.exceptions.Timeout

try:
    response = requests.get('http://some-slow-api.com', timeout=5) # 设置5秒超时
    response.raise_for_status()
except requests.exceptions.Timeout as e:
    print(f"请求超时：API响应时间过长。错误详情: {e}")
    # 考虑重试，或者给用户一个友好的提示

然后，就是HTTP状态码层面的错误。当服务器成功接收到请求，但返回了4xx（客户端错误，如404 Not Found, 401 Unauthorized）或5xx（服务器错误，如500 Internal Server Error）状态码时，

requests

response.raise_for_status()

requests.exceptions.HTTPError

try:
    response = requests.get('http://api.example.com/nonexistent-resource')
    response.raise_for_status()
    # 如果状态码是2xx，这里会继续执行
except requests.exceptions.HTTPError as e:
    print(f"HTTP错误：服务器返回了非2xx状态码。状态码: {e.response.status_code}, 错误详情: {e}")
    # 可以根据状态码进行更细致的处理，比如404可以提示资源不存在，401可以提示认证失败

最后，还有数据解析的错误。很多API返回的是JSON格式数据，如果返回的内容不是有效的JSON，那么调用

response.json()

json.JSONDecodeError

import json

try:
    response = requests.get('http://api.example.com/malformed-json')
    response.raise_for_status()
    data = response.json()
except json.JSONDecodeError as e:
    print(f"JSON解析错误：API返回了无效的JSON数据。错误详情: {e}")
    # 可能是服务器端bug，或者返回了HTML错误页面
except requests.exceptions.RequestException as e: # 捕获所有requests相关的异常
    print(f"请求发生未知错误：{e}")

我通常会把

requests.exceptions.RequestException

requests

requests

如何设计一个健壮的API请求重试机制，以应对瞬时错误？

瞬时错误，比如短暂的网络抖动、服务器负载过高导致的临时性503错误，这些都是常有的事。如果因为这些短暂的问题就直接宣告失败，那用户体验和系统稳定性都会大打折扣。所以，一个设计得当的重试机制，在我看来，是构建可靠API客户端的必备环节。

最简单的重试，可能就是用一个

while

time.sleep()

Service深入分析 WORD版

本文档主要讲述的是Service深入分析；我们还是从Service的根本意义分析入手，服务的本质就是响应客户端请求。要提供服务，就必须建立接收请求，处理请求，应答客服端的框架。我想在Android Service设计者也会无时不刻把这个服务本质框图挂在脑海中。从程序的角度，服务一定要存在一个闭合循环框架和请求处理框架。希望本文档会给有需要的朋友带来帮助；感兴趣的朋友可以过来看看

下载

我个人更推荐指数退避（Exponential Backoff）策略，并且加上抖动（Jitter）。 指数退避的核心思想是：每次重试的间隔时间呈指数级增长。比如第一次等1秒，第二次等2秒，第三次等4秒，以此类推。这样可以给服务器足够的喘息时间。 而抖动，就是在指数退避的基础上，引入一些随机性。比如，不是精确地等待4秒，而是在3秒到5秒之间随机等待。这能有效避免所有客户端在同一时刻重试，从而减轻服务器瞬时压力。

手动实现指数退避和抖动是可行的，但为了代码的简洁性和可靠性，我更倾向于使用成熟的第三方库，比如

tenacity

import requests
import logging
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 定义一个自定义的API错误，用于区分业务逻辑错误和网络/HTTP错误
class MyAPIError(Exception):
    pass

@retry(
    wait=wait_exponential(multiplier=1, min=1, max=10), # 1s, 2s, 4s, 8s, 10s (max)
    stop=stop_after_attempt(5), # 最多重试5次
    retry=retry_if_exception_type((
        requests.exceptions.ConnectionError,
        requests.exceptions.Timeout,
        requests.exceptions.HTTPError # 仅对某些HTTP错误码重试，比如5xx
    )),
    reraise=True # 如果所有重试都失败，重新抛出最后一次异常
)
def fetch_data_with_retry(url: str, params: dict = None):
    try:
        logging.info(f"尝试请求: {url} with params: {params}")
        response = requests.get(url, params=params, timeout=3)
        response.raise_for_status() # 检查HTTP状态码
        return response.json()
    except requests.exceptions.HTTPError as e:
        if 500 <= e.response.status_code < 600:
            logging.warning(f"服务器错误，尝试重试: {e.response.status_code}")
            raise # 抛出异常以触发tenacity重试
        else:
            logging.error(f"客户端或非重试型HTTP错误: {e.response.status_code}")
            raise MyAPIError(f"API返回错误: {e.response.status_code} - {e.response.text}") from e
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
        logging.warning(f"网络或超时错误，尝试重试: {e}")
        raise # 抛出异常以触发tenacity重试
    except Exception as e:
        logging.error(f"捕获到未预期的异常: {e}")
        raise MyAPIError(f"未知API错误: {e}") from e

# 示例调用
if __name__ == "__main__":
    try:
        # 假设这是一个偶尔会返回500的API
        data = fetch_data_with_retry('http://httpbin.org/status/500')
        print("成功获取数据:", data)
    except MyAPIError as e:
        print(f"最终API请求失败: {e}")
    except Exception as e:
        print(f"程序运行过程中发生未知错误: {e}")

    try:
        # 假设这是一个正常工作的API
        data = fetch_data_with_retry('http://httpbin.org/get', {'foo': 'bar'})
        print("成功获取数据:", data)
    except MyAPIError as e:
        print(f"最终API请求失败: {e}")
    except Exception as e:
        print(f"程序运行过程中发生未知错误: {e}")

在这个例子中，

@retry

ConnectionError

Timeout

HTTPError

在Python API异常处理中，日志记录和自定义异常扮演着怎样的角色？

日志记录和自定义异常，这两者在我的异常处理实践中，是相辅相成的。它们不是独立的存在，而是共同构筑起一套清晰、可维护的错误报告体系。

先说日志记录。这简直是任何生产级应用不可或缺的眼睛和耳朵。当程序在生产环境中出现问题时，我们不可能像开发时那样一步步调试。这时候，详细、准确的日志就是我们唯一的“案发现场证据”。

日志应该记录什么？

• 异常类型和错误信息：这是最基本的，知道出了什么错。
• 请求的URL和参数：在不暴露敏感信息的前提下，记录请求的上下文，能帮助我们重现问题。
• 响应状态码和内容：特别是当API返回非2xx状态码时，响应体里往往包含更详细的错误描述。
• 完整的堆栈跟踪（Traceback）：这对于定位代码中的具体位置至关重要。
• 时间戳和日志级别：便于按时间顺序分析问题，并通过日志级别（DEBUG, INFO, WARNING, ERROR, CRITICAL）区分错误的严重性。

我通常会在

try-except

ConnectionError

Timeout

WARNING

ERROR

HTTPError

ERROR

import logging
import requests

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def make_api_call(url: str, data: dict = None):
    try:
        logger.info(f"尝试请求URL: {url}, 数据: {data}")
        response = requests.post(url, json=data, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.ConnectionError as e:
        logger.error(f"API连接失败: {url} - {e}", exc_info=True) # exc_info=True 会记录完整的堆栈信息
        raise MyAPIError("无法连接到API服务") from e
    except requests.exceptions.Timeout as e:
        logger.error(f"API请求超时: {url} - {e}", exc_info=True)
        raise MyAPIError("API请求超时") from e
    except requests.exceptions.HTTPError as e:
        error_message = f"API返回错误状态码: {e.response.status_code}, URL: {url}, 响应: {e.response.text}"
        logger.error(error_message, exc_info=True)
        raise MyAPIError(error_message) from e
    except json.JSONDecodeError as e:
        logger.error(f"API响应JSON解析失败: {url} - {e}, 响应内容: {e.response.text}", exc_info=True)
        raise MyAPIError("API返回数据格式错误") from e
    except Exception as e:
        logger.critical(f"API请求发生未知严重错误: {url} - {e}", exc_info=True)
        raise MyAPIError("发生未知API错误") from e

# 注意上面代码中的MyAPIError是一个自定义异常，下面会解释

接下来是自定义异常。我个人觉得，自定义异常是抽象化、语义化错误的关键。

requests

比如，一个API可能因为用户权限不足返回403，或者因为请求参数无效返回400。虽然都是

HTTPError

HTTPError

通过定义自己的异常类，我们可以：

1. 提高代码可读性：

   raise UserPermissionDeniedError

   比

   raise HTTPError

   然后上层再检查

   if error.response.status_code == 403

   要清晰得多。
2. 简化上层处理逻辑：上层代码可以直接

   except UserPermissionDeniedError

   ，而不需要关心底层的HTTP细节。
3. 封装错误信息：自定义异常可以包含更多业务相关的错误信息，比如API返回的错误码、错误描述等，而不仅仅是HTTP状态码。

class MyAPIError(Exception):
    """自定义API请求基础异常"""
    def __init__(self, message, status_code=None, payload=None):
        super().__init__(message)
        self.status_code = status_code
        self.payload = payload

class APIAuthError(MyAPIError):
    """API认证失败异常"""
    pass

class APIValidationError(MyAPIError):
    """API请求参数验证失败异常"""
    pass

# 在 make_api_call 函数中，我们就可以这样使用：
def make_api_call_with_custom_exceptions(url: str, data: dict = None):
    try:
        # ... (同上，发起请求)
        response = requests.post(url, json=data, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401 or e.response.status_code == 403:
            logger.warning(f"API认证/权限错误: {e.response.status_code}")
            raise APIAuthError(f"认证或权限不足: {e.response.text}", status_code=e.response.status_code, payload=e.response.json()) from e
        elif e.response.status_code == 400:
            logger.warning(f"API参数验证错误: {e.response.status_code}")
            raise APIValidationError(f"请求参数无效: {e.response.text}", status_code=e.response.status_code, payload=e.response.json()) from e
        else:
            # 其他HTTP错误，抛出通用API错误
            logger.error(f"API返回未知HTTP错误: {e.response.status_code}, 响应: {e.response.text}")
            raise MyAPIError(f"API服务错误: {e.response.text}", status_code=e.response.status_code) from e
    # ... 其他异常处理保持不变

这样一来，上层调用者可以根据不同的自定义异常，执行完全不同的业务逻辑，比如

except APIAuthError:

except APIValidationError: