APScheduler失火测试

Question

我正在尝试用APScheduler测试错误触发的任务，但是在重新启动APScheduler时没有看到丢失的任务运行。我已经将APScheduler配置为：

scheduler.py

def configure_scheduler():
    jobstores = {
        'default': SQLAlchemyJobStore(url=config('DATABASE_URL'))
    }
    sched = BlockingScheduler()
    sched.configure(jobstores=jobstores)
    sched.add_job(
        test_task,
        id='test_task', 
        'interval',  
        hours=1, 
        coalesce=True, 
        max_instances=1,
        misfire_grace_time=360, 
        replace_existing=True
    )
    return sched

if __name__ == '__main__':
    scheduler = configure_scheduler()
    scheduler.start()

当我第一次启动调度器时，将test_task添加到Postgres数据库中的apscheduler_jobs表中，从启动调度程序时起，next_run_time值为1小时。然后，我试图通过以下方法来测试失火：

1. 将数据库中的next_run_time更改为当前时间
2. 等待15秒
3. 启动调度程序

当我遵循此过程时，next_run_time将再次设置为从当前时间开始的一个小时。next_run_time似乎是在SQLAlchemy作业存储的update_job方法中更新的。我看到了一个相似问题，它与持久的作业存储任务相关，而不是在失败之后运行。我看到的大多数其他 问题的解决方案是将misfire_grace_time参数添加到add_job中。我在上面的配置中尝试过这一点，但是在调度程序启动时运行错过的作业却没有什么好运气。我是不是遗漏了一些与replace_existing和misfire_grace_time参数如何交互的东西？是否需要手动检查任何作业的next_run_time是否过去，然后在启动调度程序之前运行这些作业？

我使用的是APScheduler库的v3.6.1。

对于额外的上下文，我将在Heroku上部署调度程序，并且我正在尝试处理Heroku的自动dyno循环，它每天至少发生一次。

Answer 1

在一些讨论与Alex nholm( APScheduler的创建者)在APScheduler Gitter聊天室上聊天之后，我能够确定这个工作在我的数据库中被“覆盖”了，因为我在打给add_job的电话中有replace_existing=True。这将导致调度程序在调度程序每次启动时都会替换作业存储中的作业。

我的解决方案

1. 在调度程序启动之前，请检查apscheduler_jobs表中的现有作业。
2. 对于数据库中的每个作业，根据当前时间检查next_run_time。如果next_run_time是过去的，现在运行作业。
3. 使用replace_existing=True，像以前一样安排作业。
4. 启动调度程序。