Flask全局异常处理器实现求助：线程与调度器异常捕获问题

一、主线程全局异常处理 #

sys.excepthook仅处理主线程未捕获异常，但默认会在遇到致命异常时终止程序，需修改逻辑避免主线程退出：

import sys
import traceback
import logging

logger = logging.getLogger(__name__)

def global_exception_handler(exc_type, exc_value, exc_traceback):
    # 放行正常退出指令（如sys.exit()）
    if issubclass(exc_type, SystemExit):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return
    
    # 记录完整异常栈信息（生产环境用logger替代print）
    logger.error("主线程未处理异常", exc_info=(exc_type, exc_value, exc_traceback))
    traceback.print_exception(exc_type, exc_value, exc_traceback)

sys.excepthook = global_exception_handler

二、子线程（调度器任务）异常处理 #

Python子线程的未捕获异常不会触发主线程的excepthook，需统一处理：

方法1：自定义异常处理线程类 #

import threading
import logging

logger = logging.getLogger(__name__)

class SafeThread(threading.Thread):
    def run(self):
        try:
            super().run()
        except Exception as e:
            logger.error(f"线程[{self.name}]执行异常", exc_info=True)

使用时直接替换threading.Thread为SafeThread即可。

方法2：任务函数装饰器 #

如果用APScheduler等调度器，给任务加统一捕获装饰器：

from functools import wraps
import logging

logger = logging.getLogger(__name__)

def catch_task_exceptions(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"任务[{func.__name__}]执行异常", exc_info=True)
    return wrapper

# 调度任务示例
@catch_task_exceptions
def scheduled_task():
    # 此处写任务逻辑，比如执行终端命令
    import subprocess
    result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
    print(result.stdout)

三、Flask请求上下文异常处理 #

保留原有请求异常处理器，补充日志记录：

from flask import Flask, jsonify
import logging

app = Flask(__name__)
logger = logging.getLogger(__name__)

@app.errorhandler(Exception)
def handle_request_error(error):
    logger.error("请求处理异常", exc_info=True)
    response = jsonify({'message': str(error)})
    response.status_code = 500
    return response

四、终端命令执行的异常加固 #

执行终端命令时，既要捕获调用异常，也要处理命令执行失败的情况：

import subprocess
import logging

logger = logging.getLogger(__name__)

def run_shell_command(cmd):
    try:
        # check=True会在命令返回非0码时抛出异常
        result = subprocess.run(cmd, check=True, capture_output=True, text=True)
        return result.stdout
    except subprocess.CalledProcessError as e:
        logger.error(f"命令执行失败: {' '.join(cmd)}, 错误输出: {e.stderr}", exc_info=True)
        return ""
    except Exception as e:
        logger.error(f"命令调用异常: {' '.join(cmd)}", exc_info=True)
        return ""

关键注意事项 #

• 生产环境禁用print记录日志，用Python标准logging模块配置文件/日志系统输出。
• 不要捕获KeyboardInterrupt，保留手动终止服务的入口。
• 调度器任务必须确保被异常捕获逻辑覆盖，避免单个任务崩溃影响整个调度器。

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