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 table，next_run_time 从我第一次启动启动调度程序。然后我尝试通过以下方式测试失火：

正在将我数据库中的 next_run_time 更改为当前时间
等待 15 秒
正在启动调度程序

当我执行此过程时，next_run_time 再次设置为距当前时间一小时。 next_run_time 似乎在 update_job method of the SQLAlchemy jobstore. I have seen one similar question related to persistent job store tasks not running after a misfire. The solution to most other 中更新了我看到的是将 misfire_grace_time 参数添加到 add_job。我已经按照上面的配置尝试了这个，但是没有任何运气运行在调度程序启动时错过了作业。我是否遗漏了与 replace_existing 和 misfire_grace_time 参数如何交互相关的内容？我是否需要手动检查任何作业的 next_run_time 是否在过去，然后在启动调度程序之前运行这些作业？

我正在使用 APScheduler 库的 v3.6.1。

对于其他上下文，我将在 Heroku 上部署调度程序，并且我正在尝试解决每天至少发生一次的 Heroku 的 automatic dyno cycling。

Answer 1

在与 Alex Grönholm（APScheduler 的创建者）的 APScheduler Gitter 聊天室 some discussion 之后，我能够确定该作业在我的数据库中 "overwritten" 因为我有 replace_existing=True 在我给 add_job 的电话中。这会导致调度程序在每次启动时替换作业存储中的作业。

我的解决方法

在调度程序启动之前，检查 apscheduler_jobs table 是否存在现有作业。
对于数据库中的每个作业，对照当前时间检查 next_run_time。如果next_run_time过去，运行现在的工作。
像以前一样安排作业，使用 replace_existing=True。
启动调度程序。

APScheduler 失火测试

APScheduler misfire testing

python

apscheduler