django 如何独立于后续部署运行pythonAPscheduler

ruyhziif  于 2023-06-07  发布在  Go
关注(0)|答案(1)|浏览(141)

我有一个django应用程序,它有一个python AP调度器运行一个特定的作业。我希望每周运行一次作业。可能会有后续的应用程序部署,我不希望每次发生这种情况时调度程序都重置作业。是否有任何方法可以提供过去的开始日期或检查自上次运行以来是否已经过去了一周,以确保计划程序作业独立于部署运行。
调度程序代码-

from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
scheduler.add_job(send_weekly_email, 'interval', weeks=1)
scheduler.start()
of1yzvn4

of1yzvn41#

这里的关键挑战是您的BackgroundScheduler不能跨部署/重启持久化作业信息。您可以使用APScheduler的内置作业存储,如SQLAlchemyJobStore,在不同的会话中持久化作业。但是,如果在调度最后一个作业后不到一周就进行部署,这将导致作业发送的频率高于预期,那么这种方法将无法处理这种情况。
为此,我建议单独跟踪最后发送电子邮件的日期,例如在数据库或简单文件中。然后,当作业即将执行时,它可以检查自发送最后一封电子邮件以来是否已经过去了一周。如果没有,它将跳过执行。

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from datetime import datetime, timedelta
import os

# create a scheduler with SQLAlchemyJobStore
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
scheduler = BackgroundScheduler(jobstores=jobstores)

def send_weekly_email():
    # path of the file storing the timestamp of last email sent
    last_sent_file = "/path/to/last_sent.txt"

    # check if it's time to send email again
    if os.path.exists(last_sent_file):
        with open(last_sent_file, "r") as f:
            last_sent = datetime.fromisoformat(f.read().strip())
            if datetime.now() - last_sent < timedelta(weeks=1):
                print("It's not yet time to send the email.")
                return

    # send the email here...
    
    # update the timestamp of last email sent
    with open(last_sent_file, "w") as f:
        f.write(datetime.now().isoformat())
    print("Email sent.")

# add the job
scheduler.add_job(send_weekly_email, 'interval', weeks=1)

# start the scheduler
scheduler.start()

请记住将“/path/to/last_sent.txt”替换为适合您系统的路径。
此外,考虑使用logging模块而不是print用于实际应用程序。它将帮助您跟踪何时以及为什么发送电子邮件(或不发送)。

相关问题