Django Celery 定期手动调用

Question

我正在使用 Django 和 Celery 来安排任务。

tasks.py

@periodic_task(run_every=crontab(hour='00', minute='00', day_of_week="*"))
def hello_world():
    print('hello world!)

除了每天运行该函数之外，我还需要能够手动调用它。如果用户按下前端的按钮，它将运行。我可以这样做：

@periodic_task(run_every=crontab(hour='00', minute='00', day_of_week="*"))
def hello_world():
    print('hello world!)

@task()
def hello_world():
    print('hello world!)

但这不符合 DRY。此外，我可能有多个具有相同场景的函数。我需要定期运行它，但也应要求运行。

我试过了：

def hello_world():
    print('hello world!)

@periodic_task(run_every=crontab(hour='00', minute='00', day_of_week="*"))
hello_world()

@task
hello_world()

但这不起作用。我明白了

无效语法

我无法在文档或其他 stackoverflow 答案中找到这种情况。他们谈论从 shell 调用函数。帮助将不胜感激。

谢谢！

Answer 1

你不需要定义两次任务，你的周期性任务也可以用hello_world.delay()手动调用，所以你会：

@periodic_task(run_every=crontab(hour='00', minute='00', day_of_week='*'))
def hello_world():
    print('hello world!')

def on_button_press():
    hello_world.delay()