Python API 请求中的异常处理设计-Python教程-PHP中文网

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

python api 请求中的异常处理设计

在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

登录后复制

。这在我看来，是处理API逻辑错误最直接的方式。

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

登录后复制

。这可能意味着API返回了非预期的格式，或者服务器端有错误。

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()

登录后复制

一下。但这种朴素的重试方式，很容易陷入“重试风暴”：如果所有客户端都在同一时间重试，反而会加剧服务器压力，形成恶性循环。

创客贴设计

创客贴设计，一款智能在线设计工具，设计不求人，AI助你零基础完成专业设计！

查看详情

我个人更推荐指数退避（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

登录后复制

（这里我设定为5xx系列）进行重试。如果重试次数耗尽，它会重新抛出最后一次异常，这样上层调用者就能知道最终还是失败了。这种方式，既能优雅地处理瞬时错误，又不会无限期地重试，效率和稳定性都得到了兼顾。

在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

登录后复制

，上层代码就需要解析状态码、响应体才能理解具体错误，这增加了耦合度。

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

提高代码可读性：
```
raise UserPermissionDeniedError
```
登录后复制
比
```
raise HTTPError
```
登录后复制
然后上层再检查
```
if error.response.status_code == 403
```
登录后复制
要清晰得多。
简化上层处理逻辑：上层代码可以直接
```
except UserPermissionDeniedError
```
登录后复制
，而不需要关心底层的HTTP细节。
封装错误信息：自定义异常可以包含更多业务相关的错误信息，比如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
    # ... 其他异常处理保持不变

登录后复制

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