如何让Django中的APScheduler每周任务不受部署重置影响？

方案1：使用APScheduler的持久化任务存储（推荐） #

APScheduler支持将任务状态存储到数据库中，部署重启后调度器会自动读取上次的任务执行记录，无需手动处理周期重置。

操作步骤： #

1. 安装依赖包：

pip install apscheduler django-apscheduler

2. 在Django的settings.py中注册集成app：

INSTALLED_APPS = [
    # ... 其他已有app
    'django_apscheduler',
]

3. 执行数据库迁移，创建任务存储所需的表：

python manage.py migrate

4. 修改调度器代码，配置数据库存储：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.django import DjangoJobStore

scheduler = BackgroundScheduler()
# 将Django数据库设为任务存储后端
scheduler.add_jobstore(DjangoJobStore(), 'default')

# 添加任务时指定唯一ID，避免重复创建
scheduler.add_job(
    send_weekly_email,
    'interval',
    weeks=1,
    id='weekly_email_task',
    replace_existing=True
)

scheduler.start()

部署重启后，调度器会从数据库读取任务的上次执行时间，自动延续原周期，不会重置任务。

方案2：手动记录上次执行时间 #

如果不想依赖APScheduler的持久化存储，可以自行在数据库中记录任务的上次执行时间，启动时检查并调整执行计划。

操作步骤： #

1. 创建模型存储执行日志：

from django.db import models

class TaskExecutionLog(models.Model):
    task_id = models.CharField(max_length=50, unique=True)
    last_run_time = models.DateTimeField(null=True)

    @classmethod
    def get_last_run(cls, task_id):
        try:
            return cls.objects.get(task_id=task_id).last_run_time
        except cls.DoesNotExist:
            return None

    @classmethod
    def update_last_run(cls, task_id, run_time):
        log, _ = cls.objects.get_or_create(task_id=task_id)
        log.last_run_time = run_time
        log.save()

2. 更新任务函数与调度器逻辑：

from datetime import datetime, timedelta
from apscheduler.schedulers.background import BackgroundScheduler
from .models import TaskExecutionLog

def send_weekly_email():
    # 执行邮件发送逻辑
    # ...
    
    # 记录本次执行时间
    TaskExecutionLog.update_last_run('weekly_email', datetime.now())

scheduler = BackgroundScheduler()
task_id = 'weekly_email'
last_run = TaskExecutionLog.get_last_run(task_id)

if last_run:
    time_elapsed = datetime.now() - last_run
    # 距离上次执行超过一周，立即执行一次
    if time_elapsed >= timedelta(weeks=1):
        send_weekly_email()
    
    # 计算下次执行时间，确保不早于当前时间
    next_run = last_run + timedelta(weeks=1)
    if next_run < datetime.now():
        next_run += timedelta(weeks=1)
    
    # 从计算出的时间开始，每周执行
    scheduler.add_job(
        send_weekly_email,
        'interval',
        weeks=1,
        start_date=next_run,
        id=task_id,
        replace_existing=True
    )
else:
    # 首次运行，先执行任务再启动调度
    send_weekly_email()
    scheduler.add_job(
        send_weekly_email,
        'interval',
        weeks=1,
        id=task_id,
        replace_existing=True
    )

scheduler.start()

方案3：固定每周特定时间执行（Cron表达式） #

如果不需要严格按上次执行时间间隔，而是固定每周某个时间点执行，直接用Cron表达式更简单，部署重启后会自动计算下一个固定执行时间。

示例代码：

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()
# 配置每周一上午9点执行（可按需修改day_of_week、hour、minute参数）
scheduler.add_job(
    send_weekly_email,
    'cron',
    day_of_week='mon',
    hour=9,
    id='weekly_email_task',
    replace_existing=True
)
scheduler.start()

内容的提问来源于stack exchange，提问作者VIRAJ SHAH