Hi folks,

In this post I talk about a nice tip that could come in handy while working with Celery. It is how to get all registered Celery tasks. You can use code like:

from celery import current_app
all_task_names = current_app.tasks.keys()
all_tasks = current_app.tasks.values()
foo_task = current_app.tasks['tasks.foo']

all_task_classes = [type(task) for task in current_app.tasks.itervalues()]

The task registry is only populated as the modules containing tasks are imported. If you have not imported all modules you can do like the celery worker does, and import all configured task module sources:

current_app.loader.import_default_modules()

That’s all for now. Till next time, happy software development.

References

How to get all tasks and periodic tasks in Celery [duplicate]. https://stackoverflow.com/a/12652113