Python Celery - Switch TaskRegistry implementation

Question

I'm wondering how I can change the Celery TaskRegistry implementation so that I can switch it with my own implementation. I wish to inject dependencies into tasks when they are created (e.g. when running the celery worker process).

I have tried:

from celery import Celery
Celery.registry_cls=MyCustomTaskRegistry

At the top before creating my app instance. But it does not seem to be picked up.

Any help in the right direction is greatly appreciated.

Answer 1

Found the solution myself:

app._tasks=MyCustomTaskRegistry()

Answer 2

You can define your own registry like this

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