如何从 django 前端安排任务？

Posted 2023-02-24

【中文标题】如何从 django 前端安排任务？

【英文标题】：How can I schedule a task at time from django frontend?

【发布时间】：2017-11-11 08:06:51

【问题描述】：

我正在使用 Django 编写灌溉系统，并使用 Celery 处理异步任务。

在我的应用程序中，用户选择他想要激活灌溉的日期和时间，这将存储在数据库中，他可以随时更新它们。

我知道我可以安排这样的任务，但我想更新时间。

from celery.task.schedules import crontab
from celery.decorators import periodic_task

@periodic_task(run_every=crontab(hour=7, minute=30, day_of_week="mon"))
def every_monday_morning():
    print("This runs every Monday morning at 7:30a.m.")

另一种方法是使用python-crontab

【问题讨论】：

您可以创建具有特定日期（以秒为单位）的 celery 作业。您唯一需要的是使用倒计时参数调用 apply_async。 docs.celeryproject.org/en/latest/userguide/…

可以推荐 Python APScheduler 效果很好

【参考方案1】：

通过列出我想要运行的运行时表以及何时可以从 Django 中选择来做类似的事情。

然后，定期任务本身被安排为每 30 秒运行一次，并检查此表中请求的任何内容是否准备好执行并调用其中一名工作人员。

编辑：按照格雷厄姆的要求，举个例子

在 Django 中设置节拍的设置：

CELERY_BEAT_SCHEDULE = 
    'Runtime': 
    'task': 'Runtime',
    'schedule': timedelta(seconds=15),
    ,

调用用户将通过前端更新的 Django 中定义的模型（例如，irrigation_requests）的任务：

def Runtime():
    #get all from model in state requested
    irrigation_job=Irrigation_Requests.objects.filter(Status__in=['Requested'])

    # Process each requested if flag to do just one thing then you could just test irrigation_job exists      
    for job in irrigation_job:
        ....
        print ("This runs every when requested and depending on job parameters stored in your model can have further scope")

    #update requested this could be done in loop above
    irrigation_job.update(Status='Processed')

【讨论】：

欢迎来到 ***。您能否在此处提供代码示例以提高此答案的质量？