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

Python异常处理:如何实现底层仅抛异常、上层完成计数与Discord通知?

分层解耦的核心思路

仅抛异常、上层捕获通知的本质是严格拆分各层职责,避免模块耦合:

  • 底层业务代码:只负责执行业务逻辑,碰到错误就把上下文信息封装到自定义异常里抛出,不做任何有副作用的操作——比如调外部接口、统计次数、拼接告警消息
  • 通用工具层:只实现纯逻辑工具能力,比如异常次数统计,不依赖任何外部服务、不做和自身职责无关的事
  • 顶层入口层:作为全局兜底位置,接收下层抛上来的异常,完成计数校验、告警通知这类有外部副作用的操作
对应可落地的实现示例

1. 基础自定义异常定义

首先修正原实现的基础问题:原代码里的BarryException没有继承Python内置的Exception,根本无法被正常抛出。我们需要让自定义异常支持携带业务上下文:

# lib/exceptions.py
class BarryException(Exception):
    def __init__(self, message: str, **context):
        super().__init__(message)
        # 所有传入的业务上下文(candy、color等)都存在异常实例上,供上层读取
        self.context = context

2. 剥离所有外部依赖的异常计数器

ExceptionCounter只保留异常统计、超限重抛的核心逻辑,完全去掉Discord通知相关代码:

# lib/utils.py
class ExceptionCounter:
    """
    仅负责异常次数统计,不依赖任何外部通知逻辑
    """
    def __init__(self):
        self.exception_count = {}

    def check(self, exception: type[Exception], allowed_count: int):
        exception_name = exception.__name__
        self.exception_count[exception_name] = self.exception_count.get(exception_name, 0) + 1
        # 注:原实现的判断逻辑下,如果allowed_count为0,第一次异常就会触发告警,不符合“允许连续异常x次”的设计,实际使用时可以按业务调整判断条件
        if self.exception_count[exception_name] >= allowed_count:
            raise exception(f"连续触发{exception_name}异常{self.exception_count[exception_name]}次,达到阈值{allowed_count}")

    def reset(self):
        self.exception_count.clear()

3. 底层业务代码实现

业务层只负责执行业务流程,碰到错误就把上下文和原始异常链封装到自定义异常里抛出,不碰计数器、不构造告警消息:

# 业务逻辑模块
from lib.exceptions import BarryException

candy = "hello"
color = "world"

def run_business_logic():
    try:
        int("string")
    except Exception as err:
        # 封装异常上下文和原始异常链,直接抛出
        raise BarryException(
            message="Unexpected error testing",
            candy=candy,
            color=color,
            original_error=str(err)
        ) from err

4. 顶层入口实现通知逻辑

整个程序的最入口位置,作为全局异常兜底层,集中处理计数、Discord通知逻辑:

# 程序最入口,全局异常拦截位置
import discord
from lib.exceptions import BarryException
from lib.utils import ExceptionCounter

# 全局只初始化一次计数器,避免每次新建实例导致计数失效
global_counter = ExceptionCounter()

if __name__ == "__main__":
    try:
        run_business_logic()
    except BarryException as err:
        # 接住业务层抛上来的异常,走计数校验
        try:
            global_counter.check(exception=BarryException, allowed_count=0)
        except Exception as count_trigger_err:
            # 计数达到阈值时,才构造Discord通知消息发送
            notify_payload = {
                "Title": "Unexpected Error",
                "Reason": str(count_trigger_err),
                "Payload": str(err.context)
            }
            discord.notify(notify_payload)
            # 把触发的异常继续向上抛,不阻塞原有错误处理流程
            raise count_trigger_err from err
这种设计的优势
  • 解耦所有依赖:后续如果要把Discord通知换成邮件、企业微信渠道,只要修改顶层入口的通知逻辑即可,业务代码、计数器工具类完全不需要改动
  • 逻辑边界清晰:计数器类只做纯内存操作,没有外部网络调用,测试时可以完全脱离Discord环境做单元测试
  • 避免副作用覆盖:如果在底层业务代码里直接调用Discord接口,一旦接口超时、报错,会覆盖原本的业务异常栈,反而导致问题排查困难
  • 修复原实现的计数失效问题:原代码每次在异常处理逻辑里都新建ExceptionCounter实例,exception_count字典每次都会被重置,根本统计不到连续异常次数;现在全局单例初始化计数器,统计逻辑才真正生效

内容的提问来源于stack exchange,提问作者PythonNewbie

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.10.25 05:27:01