如何在芹菜中获取所有任务和定期任务

Question

可能的重复项：
如何在 Python 中找到给定类的所有子类？

在我的 Django 项目中，我有一些 Celery Task和PeriodicTask的子类：

class CustomTask(Task):
    # stuff
class CustomPeriodicTask(PeriodicTask):
    # stuff

我需要所有任务类来添加一些自定义日志记录配置。所以我想我可以__subclasses__，但这不起作用：

>>> Task.__subclasses__()
[<unbound PeriodicTask>, <class handle_register of <Celery default:0xa1cc3cc>>]

是否有可能以某种方式以动态方式获取我所有的Task和PeriodicTask子类？

Answer 1

Celery 维护所有任务的注册表。 以下是工作人员在收到任务消息时按名称查找任务的方式：

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()]

仅当导入包含任务的模块时，才会填充任务注册表。如果您没有导入所有模块，则可以像芹菜工人一样执行，并且导入所有已配置的任务模块源：

current_app.loader.import_default_modules()

注意：import_default_modules在 Celery 2.5 之前不存在，那么您必须做：

from celery.loaders import current_loader
current_loader().init_worker()

注2：您确定要像这样编辑所有任务类吗？ 芹菜附带一个数字配置任务类的方法，例如：

CELERY_ANNOTATIONS = {
     '*': {
         'loglevel': logging.DEBUG,
         'logger': logging.getLogger('foo'),
     },
}

将设置所有任务的"日志级别"和"记录器"属性。或者，您可以使用公共基类：

class MyTask(Task):
    abstract = True   # means this base task won't be registered
    loglevel = logging.DEBUG
    logger = logging.getLogger('foo'),