基于SOLID原则重构的大型Python项目多文件错误处理最佳实践咨询

我之前维护过几个常年7×24运行的Python调度服务，太懂你现在的纠结了——既要跟着SOLID原则重构保持代码清爽，又得确保任何异常都不能搞崩整个服务，尤其是多文件场景下，try-catch放错位置要么代码乱成一团，要么漏了异常。结合你的场景，我给你拆解几个具体的实践方案：

1. 分层放置try-catch：按职责划分处理边界 #

最外层：全局兜底的主循环 #

你的调度主循环是整个服务的最后一道生命线，必须在这里加一个最顶层的try-catch，只负责防止服务崩溃，不处理具体业务异常。这样哪怕某个任务的错误处理漏了，也不会导致整个调度停摆：

import schedule
import time
import logging

logger = logging.getLogger(__name__)

# ...其他函数定义...

while True:
    try:
        schedule.run_pending()
        time.sleep(1)
    except Exception as e:
        # 这里做最关键的兜底：记录完整异常栈、发送告警（比如邮件/企业微信）
        logger.critical(f"全局未捕获异常，服务未崩溃: {str(e)}", exc_info=True)
        # 短暂休眠避免异常循环导致日志爆炸
        time.sleep(5)

任务入口层：单个调度任务的错误管家 #

每个调度任务的入口函数（比如你的f1()）是业务错误处理的第一站，这里要包揽该任务的所有错误处理逻辑，确保单个任务失败不影响其他任务，也不把异常抛到全局循环。同时符合SOLID的单一职责：核心步骤只做业务，入口负责任务生命周期管理：

def f1():
    try:
        data = f1_step1()
        # 验证失败的处理：如果用返回值就判断，用异常就捕获
        if not data_validation(data):
            logger.warning(f"f1任务数据验证失败: {str(data)}")
            return
        f1_step2()
    except ValidationError as e:
        # 捕获自定义的验证异常
        logger.warning(f"f1任务数据验证失败: {str(e)}")
    except (DBConnectionError, IOError) as e:
        # 捕获预期内的依赖服务异常（比如数据库、文件读写）
        logger.error(f"f1任务依赖服务出错: {str(e)}", exc_info=True)
    except Exception as e:
        # 捕获该任务的所有未预期异常，记录后终止任务不扩散
        logger.error(f"f1任务执行失败: {str(e)}", exc_info=True)

核心逻辑层：只抛异常不做宽泛捕获 #

像f1_step1()、f1_step2()这类核心业务函数，不要写大而全的try-catch，只在必要的地方捕获特定异常并包装成更清晰的业务异常抛出，让上层入口统一处理。这样核心代码保持干净，符合单一职责：

# 自定义业务异常，让错误更清晰
class TaskDataFetchFailedError(Exception):
    pass

def f1_step1():
    try:
        # 假设这里是数据库查询操作
        db_conn = get_db_connection()
        data = db_conn.query("SELECT critical_data FROM table")
        return data
    except DBConnectionError as e:
        # 把底层依赖异常包装成业务异常抛出
        raise TaskDataFetchFailedError(f"获取f1任务数据失败: {str(e)}") from e
    # 其他异常（比如SQL语法错误）直接抛出，由任务入口层处理

2. 验证器的错误处理：明确区分“预期失败”和“内部错误” #

你的项目有大量验证器，这里要做两种区分：

• 预期内的验证失败：比如数据格式不对、字段缺失，要么返回布尔值（在任务入口判断），要么抛出自定义的ValidationError（在入口捕获）。推荐用异常，因为能直接传递错误原因：

  class ValidationError(ValueError):
      """自定义验证异常，区分普通业务错误"""
      pass
  
  def data_validation(data):
      if not data:
          raise ValidationError("数据不能为空")
      if len(str(data.get("id", ""))) != 10:
          raise ValidationError(f"ID格式错误，必须是10位字符: {data.get('id')}")
      return True
  

• 验证器自身的异常：比如验证器依赖的配置文件读取失败、调用外部校验服务出错，这种属于验证器的内部错误，要抛出ValidationInternalError，在任务入口层用ERROR级别日志记录，而不是WARNING。

3. 贴合SOLID原则的额外细节 #

• 单一职责：每个函数只负责一件事——全局循环管服务存活，任务入口管任务错误，核心函数管业务动作，验证器管数据校验，不要让错误处理逻辑混杂在业务代码里。
• 开放封闭：用自定义异常类代替捕获通用Exception，后续新增业务场景时，只需要新增异常类，不需要修改上层的catch块，扩展更灵活。
• 依赖倒置：如果核心步骤依赖外部服务（比如数据库、API），通过抽象接口（比如DataFetcher）依赖，把依赖的异常处理封装在接口实现里，上层业务代码不需要关心具体依赖的异常类型。

4. 日志是7×24服务的核心 #

一定要确保每个catch块都记录完整的异常栈（用exc_info=True），区分日志级别：验证失败用WARNING，业务错误用ERROR，全局兜底用CRITICAL。如果有条件，把日志结构化输出（比如JSON），方便后续用监控工具检索和告警。

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