芹菜任务没有注册?

Question

我使用Django +芹菜动态调度任务。调度任务的时间将从用户数据库中查询，或者由用户动态提供。我试图传递一个参数到start_task函数，但它打破了代码，因为调度程序不能将其添加到其条目。如果没有附加参数(schedule_start_time)，它可以正常工作。．我认为这与信号有关!!

def clean_time(time):
time = time.get("created_at")
hh = time.strftime("%H")
mm = time.strftime("%M")
ss = time.strftime("%S")
dd = time.strftime("%d")
mnth = time.strftime("%m")
yy = time.strftime("%Y")
return hh, mm, ss, dd, mnth, yy

@app.on_after_configure.connect
def start_time(schedule_start_date, **kwargs):
hh, mm, ss, dd, mnth, yy = clean_time(schedule_start_date)
app.add_periodic_task(crontab((hour='{}'.format(hh), minute='{}'.format(mm), day_of_month='{}'.format(dd), month_of_year='{}'.format(mnth)), test.s("hello"))

@app.task
def test(arg):
print(arg)

Answer 1

H3ll0.

你应该使用django-芹菜-beat来管理django中的周期性芹菜任务。

这是我为测试django-celery-beat而创建的一个repo。

Periodicall回购

记住你应该像文档中描述的那样运行worker和beat。

django-celery-beat-documentation