5

重複の可能性:
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サブクラスを動的な方法で取得することは何とか可能ですか?

4

1 に答える 1

12

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_modulesCelery 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'),
于 2012-09-29T10:55:37.953 に答える