如何通过APScheduler将任务异常捕获上报至Sentry？

我来帮你搞定这个问题——你遇到的报错核心原因是对APScheduler事件对象的属性和Sentry的captureException方法的参数要求理解有误，咱们一步步来修正：

首先修复当前的监听器代码 #

你现在直接传递event.exception给sentry.captureException()，但这个方法默认需要的是sys.exc_info()返回的**(异常类型, 异常实例, 栈追踪)**元组，而event.exception只是单个异常对象，所以才会触发TypeError。

APScheduler的JobErrorEvent（也就是你收到的event）其实自带了exc_info属性，这个属性正好是符合Sentry要求的元组，修改监听函数即可：

def listen_to_exceptions(event):
    if event.exception:
        # 改用event.exc_info而非event.exception
        sentry.captureException(event.exc_info)

这样就能正确把异常和完整的栈追踪上报到Sentry了，不会再触发新的异常。

更优雅的全局异常捕获方案（适合多任务场景） #

既然你有20多个任务，不想每个都加try-except，除了事件监听器，还有两个更省心的方案：

方案1：自定义异常捕获装饰器 #

写一个通用装饰器，给所有任务函数套上，自动捕获并上报异常：

import functools
from raven import Client

sentry = Client(dsn=SENTRY_DSN)

def sentry_capture(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception:
            # 自动捕获当前异常的exc_info
            sentry.captureException()
            # 重新抛出异常，让APScheduler记录任务失败状态
            raise
    return wrapper

# 给任务函数加装饰器即可
@sentry_capture
def sample_method():
    raise ConnectionError

@sentry_capture
def another_task():
    # 你的任务逻辑
    pass

方案2：自定义执行器实现全局捕获 #

如果连装饰器都不想一个个加，可以自定义APScheduler的执行器，全局处理所有任务的异常：

from apscheduler.executors.blocking import BlockingExecutor

class SentryExecutor(BlockingExecutor):
    def run_job(self, job, job_id, store_alias, args, kwargs):
        try:
            # 调用原执行逻辑
            return super().run_job(job, job_id, store_alias, args, kwargs)
        except Exception:
            # 上报异常到Sentry
            sentry.captureException()
            # 重新抛出，不影响APScheduler的任务状态记录
            raise

# 创建调度器时使用自定义执行器
scheduler = BlockingScheduler(executors={'default': SentryExecutor()})

这样所有添加到调度器的任务，只要执行时抛出异常，都会被自动捕获并上报，完全不用修改任务代码。

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