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

Question

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

在我的Django项目中,我有一些Celery的子类,Task并且PeriodicTask:

class CustomTask(Task):
    # stuff

class CustomPeriodicTask(PeriodicTask):
    # stuff

我需要所有Task类来添加一些自定义日志记录配置.所以我认为我可以__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()]

仅在导入包含任务的模块时填充任务注册表.如果您没有像celery worker那样导入所有模块,并导入所有已配置的任务模块源:

current_app.loader.import_default_modules()

注意:import_default_modules在Celery 2.5之前不存在,那么你必须这样做:

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

注意2:您确定要编辑这样的所有任务类吗？Celery提供了许多配置任务类的方法,例如:

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

将设置所有任务的"loglevel"和"logger"属性.或者,您可以使用公共基类:

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