Сельдерей Получить список зарегистрированных задач

Ответ 1

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

Это даст словарь всех рабочих и связанных с ними зарегистрированных задач.

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

Если у вас есть несколько рабочих, выполняющих одни и те же задачи, или вам просто нужен набор всех зарегистрированных задач, он выполняет задание.

Альтернативный путь:

С терминала вы можете получить дамп зарегистрированных задач, используя эту команду

celery inspect registered

Ответ 2

В оболочке попробуйте:

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

current_app.tasks содержит все задачи, доступные в виде словаря. Ключи - все имена зарегистрированных задач в текущем приложении сельдерея, которое вы запускаете.

Ответ 3

С более новыми версиями сельдерея (4.0 и выше) правильное решение выглядит следующим образом:

 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