python - 如何在计划任务的情况下从 celery 获取任务 ID (beat)

Question

要访问 celery 任务的信息，我需要 task_id。当手动启动 celery 任务时，我可以使用 task.id 轻松获取此任务的 ID(并将其写入数据库或执行其他操作)。如果我使用定期向工作人员发送任务的 celery-beat，那似乎是不可能的。

所以我的问题是，如何在 beat 将任务发送给 celery 的 worker 时从任务中获取 id？

在工作人员收到任务的那一刻，控制台会显示任务 ID。所以我担心的是，当任务被 beat 发送给 worker 时，它没有任务 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(UTC)“提醒”celery 开始任务的时间表。

@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']))

要定期执行的任务

这是在worker收到beat的“提醒”后​​由celery执行的定时任务。

@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 来检查正在运行的 worker。它在后台使用远程控制命令。使用 i.active() 你会得到一个字典，你可以很容易地解析它。

只要我找不到任何文档可以更轻松地从周期性任务中获取 task_id，我就会坚持使用该解决方案。

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

希望对大家有帮助:)