How to get task id from celery in case of scheduled tasks (beat)

user_11538603 0
Viewed 1641

To access the info of a celery task, i need the task_id. When a celery task gets manually started i can easily get the id of this task with task.id (and the write it to DB or do something else). If i use celery-beat, which periodically sends tasks to the worker, that seems not to be possible.

So my question is, how to get the id from the task in the moment beat sends the task to celery's worker?

In the moment the worker receives the task, the console shows the task-id. So my worry is, that in the moment the task got sent by beat to the worker, it has no task id.

Manual case to get task_id:

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

task_id = task.id

Mayhaps some of you got an idea?

celerybeat celery rabbitmq python
1 Answers

i found an answer for that little issue:

If you need the task id of the tasks which was initially sent by beat, you can simply add an inspect function to your (schedulded) worker task.

Configure Periodic-Task

This is the schedule which "reminds" celery every day at 11:08 am (UTC) to kick off the task.

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

Task to be executed periodically

This is the scheduled task which will be executed by celery after the worker received the "reminder" from 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...]

So i'm making use out of app.control.inspect which lets you inspect running workers. It uses remote control commands under the hood. With i.active() you will get a dictionary, which you can easily parse.

As long as i don't find any documentation how to get the task_id from a periodic task more easier, i'll stick to that solution.

After you saved the task id you can easily poll task status etc. via AJAX for instance.

Hope that helps you guys :)

user_115386030
Related