在计划任务(BEAT)的情况下，如何从芹菜中获取任务ID

Question

要访问芹菜任务的信息，我需要task_id。当手动启动芹菜任务时，我可以轻松地使用Task.ID获得此任务的ID(并将其写入DB或做其他事情(。如果我使用定期将任务发送给工人的芹菜束，那似乎是不可能的。

所以我的问题是，当时，如何从任务中获取ID会将任务发送给芹菜的工人？

在工人收到任务的那一刻，控制台显示任务ID。因此，我担心的是，在Beat发送给工人的那一刻，它没有任务ID。

手动案例获取task_id：

task = tasks.LongRunningTask.delay(username_from_formTargetsLaden, password_from_formTargetsLaden, url_from_formTargetsLaden)
task_id = task.id

也许你们中的一些人有一个主意？

Answer 1

我找到了这个小问题的答案：

如果您需要Beat最初发送的任务的任务ID，则可以简单地将检查功能添加到(计划( worker 任务。

配置定期任务

这是每天在11:08 AM(UTC("提醒"芹菜开始任务的时间表。

@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    test = sender.add_periodic_task(crontab(minute=8, hour=11), CheckLists.s(app.config['USR'], app.config['PWD']))

要定期执行的任务

这是计划的任务，该任务将由芹菜执行，在工人收到Beat的"提醒"。

@celery.task(bind=True)
def CheckLists(self, arg1, arg2):
    #get task_id von scheduled Task 'Check-List'
    i = inspect()
    activetasks = i.active()
    list_of_tasks = {'activetasks': activetasks}
    task_id = list_of_tasks['activetasks']['celery@DESKTOP-XXXXX'][0]['id']   #adapt this section depending on environment (local, webserver, etc...)
    task_type = "CHECK_LISTS"
    task_id_to_db = Tasks(task_id, task_type)
    db.session.add(task_id_to_db)
    db.session.commit()
    long_runnning_task 
    [...more task relevant code here...] 

因此，我可以从app.control.inspect中使用，这使您可以检查跑步工人。它使用引擎盖下的遥控命令。使用i.active()，您将获得一个字典，您可以轻松解析。

只要我找不到任何文档如何更容易从周期性任务中获取task_id，我就坚持使用该解决方案。

保存任务ID后，您可以轻松地通过AJAX进行轮询任务状态等。

希望可以帮助你们：(