Python多外部API调用场景下Try-Except代码块重构方案咨询

我之前维护过对接多个第三方API的服务，完全懂你这种重复写try-except块的痛苦——不仅代码冗余，后续改逻辑（比如加失败任务入队）还要挨个改函数，太麻烦了。下面的方案既能消除冗余，又能保留关键字参数的可读性，还能轻松扩展未来的需求：

1. 核心方案：通用异常处理装饰器 #

装饰器是处理这种批量函数异常逻辑的最佳选择，配合functools.wraps可以完整保留原函数的签名（包括关键字参数的IDE提示），完全解决你之前封装execute_func丢失可读性的问题。

第一步：定义通用装饰器 #

import functools
from typing import Callable, Any

# 假设你有自定义异常类，没有的话可以先定义
class SlackAPIError(Exception):
    pass

def handle_api_exceptions(failure_handler: Callable = None):
    """API调用通用异常处理装饰器，支持自定义失败处理逻辑（比如入队）"""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)  # 保留原函数的签名和文档字符串，不影响参数可读性
        def wrapper(*args, **kwargs) -> Any:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                # 基础异常处理：记录日志+抛出自定义异常
                error_msg = f"调用{func.__name__}失败: {str(e)}"
                print(error_msg)  # 替换为你的日志系统即可
                # 执行自定义失败逻辑（比如未来的失败任务入队）
                if failure_handler:
                    failure_handler(func.__name__, args, kwargs, e)
                # 可根据业务选择返回错误字典或抛出自定义异常
                raise SlackAPIError(error_msg) from e
        return wrapper
    return decorator

第二步：应用到你的API函数 #

原来的slack_client.py里的函数直接加装饰器即可，完全不用修改内部逻辑，关键字参数的可读性完全保留：

# slack_client.py
from my_decorators import handle_api_exceptions

# 提前定义失败任务入队逻辑（未来需要时直接启用）
def enqueue_failed_task(func_name: str, args: tuple, kwargs: dict, exception: Exception):
    # 这里写你的任务入队逻辑，比如放到Celery/Redis队列
    print(f"将失败任务{func_name}入队: 参数={kwargs}, 错误={str(exception)}")

# 应用装饰器，需要入队就传failure_handler，不需要就留空
@handle_api_exceptions(failure_handler=enqueue_failed_task)
def create_space(name: str, team_id: str, is_private: bool = False):
    # 原有的API调用逻辑，比如调用Slack官方API
    response = slack_api.post("spaces.create", json={"name": name, "team_id": team_id, "private": is_private})
    response.raise_for_status()
    return response.json()

@handle_api_exceptions(failure_handler=enqueue_failed_task)
def delete_space(space_id: str):
    # 原有的API删除逻辑
    response = slack_api.delete(f"spaces/{space_id}")
    response.raise_for_status()
    return response.json()

调用时完全和原来一样，IDE会正常提示关键字参数，可读性拉满：

# 和重构前的调用方式毫无区别，参数提示正常
create_space(name="研发部专属空间", team_id="T12345", is_private=True)
delete_space(space_id="S67890")

2. 灵活扩展：适配不同需求 #

• 针对特定异常捕获：如果某些API需要捕获特定异常（比如requests.exceptions.HTTPError），可以给装饰器加参数定制：

  def handle_api_exceptions(allowed_exceptions=(Exception,), failure_handler=None):
      def decorator(func):
          @functools.wraps(func)
          def wrapper(*args, **kwargs):
              try:
                  return func(*args, **kwargs)
              except allowed_exceptions as e:
                  # 后续逻辑不变
                  ...
          return wrapper
      return decorator
  

• 动态切换失败逻辑：比如测试环境不需要入队，生产环境需要，只需通过配置动态传入failure_handler，不用修改函数代码。

3. 替代方案：上下文管理器（适合临时代码块） #

如果你偶尔需要在某个代码块（不是整个函数）处理异常，可以用上下文管理器，但装饰器更适合批量函数的场景：

from contextlib import contextmanager

@contextmanager
def api_exception_handler(func_name: str, failure_handler=None):
    try:
        yield
    except Exception as e:
        error_msg = f"调用{func_name}失败: {str(e)}"
        print(error_msg)
        if failure_handler:
            failure_handler(func_name, (), {}, e)
        raise SlackAPIError(error_msg) from e

# 使用方式
with api_exception_handler("create_space", enqueue_failed_task):
    response = slack_api.post(...)

重构前后对比 #

• 重构前：每个函数都要写重复的try-except块，修改异常逻辑要挨个改函数。
• 重构后：异常逻辑集中在装饰器里，所有API函数只需加一行装饰器，后续加失败入队、修改异常处理逻辑都只需调整装饰器或失败处理函数，维护成本大幅降低。

内容的提问来源于stack exchange，提问作者Victor