Python Celery - Switch TaskRegistry implementation

Question

我想知道如何更改Celery TaskRegistry实现，以便使用自己的实现进行切换。我希望在创建任务时（例如，在运行芹菜工作进程时）将依赖项注入到任务中。

我试过：

from celery import Celery
Celery.registry_cls=MyCustomTaskRegistry

在创建我的应用程序实例之前的顶部。但它似乎没有被拾起。

我们非常感谢在正确方向上提供的任何帮助。

Answer 1

您可以像这个一样定义自己的注册表

from celery import Celery
a = Celery()
print(a.registry_cls)    # gives original registry
from celery.app.registry import TaskRegistry
class MyCustomRegistry(TaskRegistry):
    pass
a.registry_cls = MyCustomRegistry
print(a.registry_cls)     # gives your custom registry

Answer 2

自己找到了解决方案：

app._tasks=MyCustomTaskRegistry()