[question] inject in celery tasks

[Martin25699]

Hi, what are the best practices for inject in celery tasks?

from celery import shared_task

class FooService:
    @inject
    def __init__(bar: Bar):
          self.bar = bar

    def run_method()
        pass

@shared_task(name='foo_task')
def foo_task(service: FooService)
    service.run_method()

How to implement correctly?

[jstasiak]

I've never used Celery directly, but it most likely will need an integration layer like https://github.com/alecthomas/flask_injector or https://github.com/blubber/django_injector. I don't know if Celery is extensible in this regard, so there may be a bit of hacking involved.

[proofit404]

Hi,

I'm the author of an alternative dependency injector library (dependencies part of the dry-python project).

We already have integration with Celery.

We decide not to fight with Celery worker bootstrap but provide a way to register a task with injector outside of the Celery task function.

If anyone is interested to backport this to the project, I invite you to take a look at our implementation.

Best regards, Artem.

[davidparsson]

If I remember correctly, class based Celery tasks (as in the code linked above) are kind of like singletons, so a single instance of a task class may be used for several runs.

This was not acceptable for me so I built a custom task decorator on top of Celery's decorator. Mine gets an injector instance that has been set in the Celery config, which is then used to call the task. It's a bit hacky but it works.

[proofit404]

Actually, we use function-based and class-based tasks depending on the bind parameter in the scope of the Injector.

https://github.com/dry-python/dependencies/blob/5fcc9313da03f28af2165aa9cb434aa333d0f1d9/src/dependencies/contrib/_celery.py#L55-L63

[davidparsson]

I missed that, sorry, and thanks!