Python脚本如何捕获全局异常报错

wen python案例 29

本文目录导读:

Python脚本如何捕获全局异常报错

  1. 使用 sys.excepthook 钩子函数
  2. 使用 threading.excepthook(线程异常处理)
  3. 装饰器方式捕获函数异常
  4. 使用 atexit 模块处理程序退出时的异常
  5. 上下文管理器方式
  6. Web框架中的全局异常处理(以Flask为例)
  7. 综合实践:生产环境推荐的方案
  8. 注意事项

在Python中捕获全局异常报错有多种方法,以下是几种常用的实现方式:

使用 sys.excepthook 钩子函数

import sys
import traceback
def global_exception_handler(exc_type, exc_value, exc_traceback):
    """全局异常处理函数"""
    # 记录异常信息到文件
    with open('error.log', 'a', encoding='utf-8') as f:
        f.write(f"异常类型: {exc_type.__name__}\n")
        f.write(f"异常值: {exc_value}\n")
        f.write("追踪信息:\n")
        traceback.print_tb(exc_traceback, file=f)
        f.write("-" * 50 + "\n")
    # 打印到控制台
    print(f"发生未捕获异常: {exc_value}")
    # 阻止默认处理(可选)
    # sys.exit(1)
# 设置全局异常处理器
sys.excepthook = global_exception_handler
# 测试代码
def test_function():
    raise ValueError("测试异常")
test_function()

使用 threading.excepthook(线程异常处理)

import threading
import sys
def thread_exception_handler(args):
    """线程异常处理函数"""
    print(f"线程 {args.thread.name} 发生异常:")
    print(f"类型: {args.exc_type}")
    print(f"值: {args.exc_value}")
    # 记录日志
    with open('thread_errors.log', 'a') as f:
        f.write(f"Thread: {args.thread.name}, Error: {args.exc_value}\n")
# 设置线程异常处理器
threading.excepthook = thread_exception_handler
def worker():
    raise RuntimeError("线程中的异常")
# 测试
t = threading.Thread(target=worker)
t.start()
t.join()

装饰器方式捕获函数异常

import functools
import traceback
def catch_exceptions(func):
    """装饰器:捕获函数中的所有异常"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print(f"函数 {func.__name__} 发生异常:")
            print(f"错误信息: {e}")
            traceback.print_exc()
            # 记录到文件
            with open('function_errors.log', 'a') as f:
                f.write(f"Function: {func.__name__}\n")
                traceback.print_exc(file=f)
                f.write("\n")
    return wrapper
@catch_exceptions
def risky_function(x):
    return 1 / x
risky_function(0)  # 会被捕获

使用 atexit 模块处理程序退出时的异常

import atexit
import sys
def cleanup_handler():
    """程序退出时的清理处理"""
    print("程序正在退出...")
    # 检查是否有未处理的异常
    if sys.exc_info()[0] is not None:
        print(f"程序异常退出: {sys.exc_info()[1]}")
atexit.register(cleanup_handler)
# 生产环境中更好的方式:注册多个清理函数
@atexit.register
def another_cleanup():
    print("执行另一个清理操作")

上下文管理器方式

class GlobalExceptionCatcher:
    """上下文管理器:在代码块中捕获所有异常"""
    def __enter__(self):
        self.old_excepthook = sys.excepthook
        sys.excepthook = self.handler
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        sys.excepthook = self.old_excepthook
        return False  # 不抑制异常
    @staticmethod
    def handler(exc_type, exc_value, exc_traceback):
        print(f"在上下文管理器中捕获: {exc_value}")
        # 处理异常...
# 使用
with GlobalExceptionCatcher():
    # 这里的异常会被捕获
    raise ValueError("测试")

Web框架中的全局异常处理(以Flask为例)

from flask import Flask, jsonify
app = Flask(__name__)
@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "资源不存在"}), 404
@app.errorhandler(500)
def internal_error(error):
    return jsonify({"error": "服务器内部错误"}), 500
# 捕获所有未处理异常
@app.errorhandler(Exception)
def unhandled_exception(error):
    app.logger.error(f'未处理异常: {error}')
    return jsonify({"error": "未知错误"}), 500

综合实践:生产环境推荐的方案

import sys
import traceback
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
    level=logging.ERROR,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('production_errors.log'),
        logging.StreamHandler()
    ]
)
def production_exception_handler(exc_type, exc_value, exc_traceback):
    """生产环境全局异常处理器"""
    # 记录到日志
    logging.error(
        "未捕获异常",
        exc_info=(exc_type, exc_value, exc_traceback)
    )
    # 发送告警(示例)
    send_alert(f"关键错误: {exc_value}")
    # 保存崩溃信息
    save_crash_report(exc_type, exc_value, exc_traceback)
    # 程序退出
    sys.exit(1)
def send_alert(message):
    """发送告警通知(示例)"""
    print(f"发送告警: {message}")
def save_crash_report(exc_type, exc_value, exc_traceback):
    """保存崩溃报告"""
    with open(f'crash_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log', 'w') as f:
        traceback.print_exception(exc_type, exc_value, exc_traceback, file=f)
# 设置全局处理器
sys.excepthook = production_exception_handler
# 测试
raise ValueError("测试生产环境异常处理")

注意事项

import signal
# 处理信号中断
def signal_handler(signum, frame):
    print(f"收到信号 {signum}")
    sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)  # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # 终止信号
# 主程序逻辑
if __name__ == "__main__":
    try:
        # 主要业务代码
        raise RuntimeError("主要程序错误")
    except Exception as e:
        print(f"主程序捕获异常: {e}")
        # 执行清理操作
        cleanup()

最佳实践建议:

  1. 开发环境:使用 sys.excepthook 记录详细追踪信息
  2. 生产环境:记录日志 + 发送告警
  3. 多线程程序:同时设置 threading.excepthook
  4. Web应用:使用框架提供的异常处理机制
  5. 关键系统:考虑使用专业错误跟踪服务(如Sentry)

抱歉,评论功能暂时关闭!