Celery Get List Of Registered Tasks

Question

Is there a way to get a list of registered tasks?

I tried:

celery_app.tasks.keys()

Which only returns built in Celery tasks like celery.chord, celery.chain etc.

have you checked out this so thread? http://stackoverflow.com/questions/12651872/how-to-get-all-tasks-and-periodic-tasks-in-celery — VF_, Oct 02 '14 at 12:34

Chillar Anand · Answer 1 · 2021-03-23T03:54:05.323

67

For the newer versions of celery(4.0 or above), we can get registered tasks as follows.

from celery import current_app 
tasks = current_app.tasks.keys()

For older versions of celery, celery < 4, we can get registered tasks as follows.

from celery.task.control import  inspect
i = inspect()
i.registered_tasks()

This will give a dictionary of all workers & related registered tasks.

from itertools import chain
set(chain.from_iterable( i.registered_tasks().values() ))

In case if you have multiple workers running same tasks or if you just need a set of all registered tasks, it does the job.

Alternate Way:

From terminal you can get a dump of registered tasks by using this command

celery inspect registered

To inspect tasks related to a specific app, you can pass app name

celery -A app_name inspect registered

edited Mar 23 '21 at 03:54

answered Oct 06 '14 at 06:55

Chillar Anand

27,936
9
119
136

1

Whic doesn't list all the registered tasks for some reason. – Richard Knop Oct 07 '14 at 13:33
Thats weird. Not sure what's going wrong. Can you share your celery config and your tasks? – Chillar Anand Oct 07 '14 at 14:23
29

for anyone who is using Celery with Django, `celery -A {celery_app_name} inspect registered` does the trick. – Iman Akbari Jun 28 '16 at 16:02
1

@RichardKnop The problem is that all the inspection functions of Celery are based on a [broadcast](http://docs.celeryproject.org/en/latest/reference/celery.app.control.html#celery.app.control.Control.broadcast) method that times out if it does not receive a reply within a specific time frame. If a worker is very busy, it may not reply within the timeout, and the result you'll get from the inspection call will be as if this worker did not exist at all. (The default timeout is 1 second.) – Louis Jul 06 '16 at 18:04
This isn't a reliable solution, especially for testing. `i.registered_tasks()` returns None during unittests. – Cerin Mar 22 '21 at 19:35

score 26 · Answer 2 · edited Jul 18 '23 at 19:29

26

With the newer versions of celery ( 4.0 and above ), the following seems to be the right way:

from celery import current_app

current_app.loader.import_default_modules()

tasks = list(sorted(name for name in current_app.tasks
                            if not name.startswith('celery.')))
return tasks

edited Jul 18 '23 at 19:29

palamunder

2,555
1
19
20

answered Sep 13 '19 at 10:44

Navid Khan

979
11
24

2

Thanks. This worked perfectly for me, whereas the accepted answer does not. – Cerin Mar 22 '21 at 19:46
1

Worked for me. Thank you – hamza saber Jul 15 '21 at 00:29
2

I think this returns a list of tasks(names) as strings but is it possible to return them as objects in order to access their parameters? – user3821178 Feb 28 '22 at 01:01

score 11 · Answer 3 · edited Sep 09 '19 at 14:11

11

In a shell, try:

from celery import current_app 
print(current_app.tasks.keys())

current_app.tasks has all tasks available as a dictionary. The keys are all the names of registered tasks in the current celery app you are running.

edited Sep 09 '19 at 14:11

Scott Stafford

43,764
28
129
177

answered Mar 19 '19 at 15:07

Mike - kentivo

151
1
3

What could be the reason my task is not showing through python, but it does through CLI `celery worker inspect registered`? – Constantin Jan 08 '22 at 12:40

Celery Get List Of Registered Tasks

3 Answers3

Linked