如何获得芹菜所有任务和周期性任务

可能重复：
How can I find all subclasses of a given class in Python?如何获得芹菜所有任务和周期性任务

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

class CustomTask(Task): 
    # stuff 

class CustomPeriodicTask(PeriodicTask): 
    # stuff

我需要所有任务类来添加一些自定义日志配置通货膨胀。所以我想我能给我们__subclasses__，但这不起作用：

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

是它在某种程度上可能得到以动态的方式我所有的Task和PeriodicTask子？

来源

2012-09-29 Martin

我认为，这不是，因为其他的问题是围绕着'__subclass__'属性的有效使用，有什么不能在这里工作，你可以看到在这个问题。 – Martin

芹菜维护所有任务的注册表。这是当它接收任务消息的工作如何按名称查询一个任务：

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不存在芹菜2.5之前，那时你不得不到做：

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

NOTE2：你确定你要编辑的所有任务类这样吗？芹菜配有多种方式来配置任务类，例如：

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'),

来源

2012-09-29 10:55:37 asksol

我想为每个任务设置单个记录器，但使用相同的设置，因此全局记录器将不适合此处。另一方面，我不想为每个任务配置它，所以我想到了一个程序化的方法。 – Martin

您可以使用基类或注释为每个任务创建一个单一的记录器：class anno（object）：def annotate（self，task）：return {'logger'：logging.getLogger（'tasks。％s'％任务名称）; CELERY_ANNOTATIONS = [Anno（）]' – asksol

如何获得芹菜所有任务和周期性任务

回答

相关问题