python - 重启 Wsgi 服务器时重启 ApScheduler Python 中的作业

Question

我正在使用 python Apscheduler 来安排我的工作。我所有的作业都存储为 cron 作业并使用 BackgroundScheduler。我有以下代码:

def letschedule():
    jobstores = {
        'default': SQLAlchemyJobStore(url=app_jobs_store)
    }
    executors = {
        'default': ThreadPoolExecutor(20),
        'processpool': ProcessPoolExecutor(5)
    }
    job_defaults = {
        'coalesce': False,
        'max_instances': 1,
        'misfire_grace_time':1200
    }
    scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
    #jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc
    return scheduler

然后我在应用程序中启 Action 业调度程序，如下所示:

sch = letschedule()
sch.start()
log.info('the scheduler started')

我有以下添加工作功能。

def addjobs():


    jobs = []
    try:
        sch.add_job(forecast_jobs, 'cron', day_of_week=os.environ.get("FORECAST_WEEKOFDAY"),  
            id="forecast",
             replace_existing=False,week='1-53',hour=os.environ.get("FORECAST_HOUR"), 
             minute=os.environ.get("FORECAST_MINUTE"), timezone='UTC')
        jobs.append({'job_id':'forecast', 'type':'weekly'})

        log.info('the forecast added to the scheduler')
    except BaseException as e:
        log.info(e)
        pass
    try:
        sch.add_job(insertcwhstock, 'cron', 
            id="cwhstock_data", day_of_week='0-6', replace_existing=False,hour=os.environ.get("CWHSTOCK_HOUR"),
            minute=os.environ.get("CWHSTOCK_MINUTE"),
            week='1-53',timezone='UTC')
        jobs.append({'job_id':'cwhstock_data', 'type':'daily'})
        log.info('the cwhstock job added to the scheduler')
    except BaseException as e:
        log.info(e)
        pass
    return json.dumps({'data':jobs})

我在 flask 应用程序中使用它，我调用/activatejobs 并将作业添加到调度程序中，它工作正常。但是，当我重新启动 wsgi 服务器时，作业不会再次启动，我必须删除 .sqlite 文件并再次添加作业。我想要的是作业应该在调度程序启动后自动重新启动(如果数据库中已经有作业。) 我试图通过一些方法得到这样的结果，但做不到。任何帮助将不胜感激。提前致谢。

Answer 1

我在使用 FastApi 框架时也遇到了同样的问题。将此代码添加到我的 app.py 后，我可以解决问题:

scheduler = BackgroundScheduler()
pg_job_store = SQLAlchemyJobStore(engine=my_engine)
scheduler.add_jobstore(jobstore=pg_job_store, alias='sqlalchemy')
scheduler.start()

添加这段代码，在我重新启动应用程序服务器后，我可以看到 apscheduler 日志搜索作业:

2021-10-20 14:37:53,433 - apscheduler.scheduler - INFO     => Scheduler started
2021-10-20 14:37:53,433 - apscheduler.scheduler - DEBUG    => Looking for jobs to run
Jobstore default:
    No scheduled jobs
Jobstore sqlalchemy:
    remove_from_db_job (trigger: date[2021-10-20 14:38:00 -03], next run at: 2021-10-20 14:38:00 -03)
2021-10-20 14:37:53,443 - apscheduler.scheduler - DEBUG    => Next wakeup is due at 2021-10-20 14:38:00-03:00 (in 6.565892 seconds)

它对我有用。