如何获取Celery Beat中PeriodicTask的下次执行时间？

要计算你创建的PeriodicTask的下次执行时间，需要结合任务的运行记录和间隔配置推导，具体逻辑如下：

核心计算逻辑 #

PeriodicTask的下次执行时间依赖两个关键属性：

• last_run_at：任务最后一次运行的时间（未运行过则为None）
• interval：关联的IntervalSchedule实例，包含every（间隔数值）和period（间隔单位，如seconds/minutes/hours）

代码实现 #

from datetime import timedelta

# 假设已获取创建好的task对象
# 根据间隔配置构建时间差
interval_kwargs = {task.interval.period: task.interval.every}
delta = timedelta(**interval_kwargs)

# 计算基础下次执行时间
if task.last_run_at is None:
    # 任务从未运行过，以创建时间为起点计算
    next_run_time = task.created + delta
else:
    # 任务已运行过，以上次运行时间为起点计算
    next_run_time = task.last_run_at + delta

# 若任务设置了start_time，确保下次执行时间不早于该时间
if task.start_time and next_run_time < task.start_time:
    next_run_time = task.start_time

# 可按需格式化输出
print(next_run_time.strftime("%Y-%m-%d %H:%M:%S"))

补充说明 #

• 如果任务使用CrontabSchedule而非IntervalSchedule，需解析crontab表达式计算下次执行时间，可借助celery.schedules.crontab的is_due()方法获取间隔时长后再计算。
• django-celery-beat的PeriodicTask只有在实际被Celery调度执行后，才会自动更新last_run_at字段。

内容的提问来源于stack exchange，提问作者Sherlock Holmes