如何在Rocketry中运行时延迟任务下次执行时间（不修改原调度）

该需求完全可行，Rocketry支持运行时修改任务的下次执行时间，且不会破坏原有的调度规则，后续执行仍会基于修改后的时间按原规则推进。

实现步骤 #

• 获取目标任务实例：通过Rocketry的session获取指定名称的任务对象
• 计算新的下次执行时间：根据需求确定延迟后的时间点（可基于当前时间或原计划时间）
• 修改任务的next_run属性：直接赋值更新任务的下次执行时间

代码示例 #

基础任务定义 #

from rocketry import Rocketry
from rocketry.conds import every
from datetime import datetime, timedelta

def func():
    print(f"任务执行于: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

app_rocketry = Rocketry()

# 创建10个复用同一函数、不同调度规则的任务
for i in range(10, 20):
    app_rocketry.task(name=f"func{i}", func=func, start_cond=every(f"{i**2} minutes"))

延迟任务下次执行的函数 #

def delay_task_next_run(task_name, delay_minutes):
    # 获取指定任务
    task = app_rocketry.session.get_task(task_name)
    if not task:
        print(f"未找到任务: {task_name}")
        return
    
    # 方式1：基于原计划的下次执行时间延迟（比如原计划12:00，延迟8分钟到12:08）
    # original_next = task.next_run
    # new_next_run = original_next + timedelta(minutes=delay_minutes)
    
    # 方式2：基于当前时间延迟（符合你示例中12:03触发，延迟到12:08的场景）
    new_next_run = datetime.now() + timedelta(minutes=delay_minutes)
    
    # 更新下次执行时间
    task.next_run = new_next_run
    print(f"任务 {task_name} 的下次执行时间已更新为: {new_next_run.strftime('%Y-%m-%d %H:%M:%S')}")

运行时调用示例 #

# 假设在12:03时调用，将func10的下次执行时间延迟5分钟到12:08
delay_task_next_run("func10", 5)

关键说明 #

• 修改next_run后，任务会在指定时间执行；执行完成后，Rocketry会自动根据原调度规则（如every 5 minutes）从新的执行时间开始计算下一次执行点，确保后续仍按原周期运行。
• 若使用时区，需确保new_next_run是带时区信息的datetime对象，避免时间不一致问题。

内容的提问来源于stack exchange，提问作者Hossein Gholami