如何使用Scheduler库获取Python定时作业的执行状态？

刚好我对这个scheduler库有了解，给你几个可行的方案来实现任务执行状态（成功/失败）的追踪：

方案1：包装任务函数，主动捕获异常并记录状态 #

这个方法最直接，我们可以给任务函数加一个装饰器，在函数执行时捕获异常，同时记录执行状态到一个全局字典或者其他存储结构里：

import datetime as dt
import time
from scheduler import Scheduler

# 用字典存储任务的最新执行状态，键可以用函数名作为唯一标识
job_status_tracker = {}

def track_status(func):
    def wrapper(*args, **kwargs):
        job_name = func.__name__
        try:
            # 执行原任务函数
            func(*args, **kwargs)
            # 执行成功，标记状态为Success
            job_status_tracker[job_name] = "Success"
        except Exception as e:
            # 执行失败，记录失败原因
            job_status_tracker[job_name] = f"Failed: {str(e)}"
            # 如果需要让scheduler感知到异常（比如停止重试），可以取消下面的raise
            # raise
    return wrapper

# 给目标任务函数加上装饰器
@track_status
def foo():
    print("foo")
    # 测试失败场景可以取消下面的注释
    # raise ValueError("模拟任务执行出错")

schedule = Scheduler()
schedule.minutely(dt.time(second=15), foo)

while True:
    schedule.exec_jobs()
    # 按需打印或使用任务状态
    print(f"当前任务状态: {job_status_tracker}")
    time.sleep(1)

方案2：利用Job对象的属性结合异常捕获 #

当你调用schedule.minutely()时，会返回一个Job实例，这个实例自带attempts属性（已执行次数）。我们可以通过对比任务执行前后的attempts数值，结合异常捕获来判断执行状态：

import datetime as dt
import time
from scheduler import Scheduler

def foo():
    print("foo")
    # 测试失败场景取消注释
    # raise TypeError("任务执行异常")

schedule = Scheduler()
# 保存返回的Job实例，用于后续追踪
foo_job = schedule.minutely(dt.time(second=15), foo)

while True:
    prev_attempts = foo_job.attempts
    try:
        schedule.exec_jobs()
        # 如果执行后attempts增加，说明任务被触发且执行成功
        if foo_job.attempts > prev_attempts:
            print("foo任务执行成功！")
    except Exception as e:
        # 如果任务抛出未被捕获的异常，会在这里捕获到
        print(f"foo任务执行失败: {str(e)}")
    time.sleep(1)

补充说明 #

• scheduler库的exec_jobs()方法默认会捕获任务执行时的大部分异常，并自动增加Job实例的attempts计数，但不会主动对外暴露执行状态，所以需要我们自己封装逻辑。
• 如果需要记录每次执行的详细状态（比如执行时间、状态详情），可以把job_status_tracker改成列表，每次执行时追加一条包含时间戳和状态的记录。

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