You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python如何捕获CustomWarning自定义警告类且不中断代码执行?

实现方案

核心问题说明:Python 中默认抛出任何异常都会直接中断当前执行流,所以你原本设想的在外层try-except捕获CustomWarning的逻辑无法满足「不中断handle_request执行」的要求,异常一旦抛出到外层try块,handle_request就已经提前终止了。我们需要调整实现逻辑,先收集所有警告,等handle_request执行完成后再统一处理。

方案1:自行维护警告列表(改造成本最低)

步骤1:定义CustomWarning类

class CustomWarning(Warning):
    def __init__(self, type_, header, warning_message):
        self.type_ = type_
        self.header = header
        self.warning_message = warning_message
        super().__init__(warning_message)

步骤2:给处理类新增警告存储和触发逻辑

在你的服务类初始化时新增警告存储容器:

class YourService:
    def __init__(self):
        # 原有初始化逻辑
        self.warnings = []

新增警告触发方法,替代直接raise的写法:

def add_warning(self, warning: CustomWarning):
    self.warnings.append(warning)

步骤3:调整辅助函数的警告触发逻辑

你可以保留和错误判断几乎一致的写法,只是把raise换成调用add_warning:

warning_required = is_warning_required(column_id, reference)
if warning_required:
    self.add_warning(CustomWarning(
        type_="reference_warn",
        header="引用警告",
        warning_message="当前引用可能导致后续计算结果不符合预期"
    ))

如果你一定要保留raise的写法,只需要在每个辅助函数内部包裹一层try-except捕获CustomWarning,存入列表后继续执行即可:

def your_helper_func(self, column_id, reference):
    try:
        # 原有辅助函数逻辑
        warning_required = is_warning_required(column_id, reference)
        if warning_required:
            raise CustomWarning(...)
        # 剩余逻辑正常执行
    except CustomWarning as w:
        self.warnings.append(w)

步骤4:改造receive_message逻辑

try:
    self.warnings = [] # 每次新请求先清空历史警告
    self.handle_request(event)

    # 先处理所有收集到的警告
    for w in self.warnings:
        response = {
            'event': 'edit_warning',
            'id': event['id'],
            'type': w.type_,
            'header': w.header,
            'warning_message': w.warning_message
        }
        self.send(response)

    # 再发送成功响应
    self.send({success: True})
         
except CustomError as e:
    print(get_recent_traceback())
    print(e)
    # 如果需要出错时也返回之前收集的警告,可以在这里加警告发送逻辑
    response = {
        'event': 'edit_error',
        'id': event['id'],
        'type': e.type_,
        'header': e.header,
        'to_fix': e.to_fix
    }
    self.send(response)

方案2:使用Python标准库warnings模块(更符合规范)

你也可以直接复用Python原生的警告处理体系,不需要自己维护警告列表:

  1. 定义CustomWarning继承自Warning类
  2. 辅助函数中触发警告用warnings.warn(CustomWarning(...))
  3. 在receive_message中用warnings.catch_warnings上下文管理器收集所有请求过程中产生的警告,执行完成后统一处理即可。

两种方案都可以完全满足你的需求:错误触发时立即中断执行返回错误,警告触发时不中断执行,等所有逻辑跑完后再返回所有警告信息。

内容的提问来源于stack exchange,提问作者Aaron Diamond-Reivich

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.29 21:54:04