Использование методов класса в качестве задач сельдерея

Я пытаюсь использовать методы класса как задачи django-celery, отмечая его с помощью декоратора @task. Такая же ситуация описывается здесь , по словам Ананда Джеяхара. Это что-то вроде этого

class A:
    @task
    def foo(self, bar):
        ...

def main():
    a = A()
    ...
    # what i need
    a.foo.delay(bar) # executes as celery task 
    a.foo(bar) # executes locally

Проблема даже в том, что я использую экземпляр класса, подобный этому a.foo.delay(bar), он говорит, что foo требуется как минимум два аргумента, что означает, что указатель self отсутствует.

Дополнительная информация:

Я не могу преобразовать класс в модуль из-за наследования
Методы сильно зависят от членов класса, поэтому я не могу сделать их статическими
Класс маркировки как задачи с помощью декоратора @task делает класс задачей сам по себе, и можно было бы выполнить методы из метода run(), используя некоторый аргумент в качестве ключа для выбора метода, но это не совсем то, что я хотите.
Создание экземпляра класса и передача его в качестве аргумента self для методов изменяет способ, которым я выполняю методы не как celery taks, а как обычные методы (т.е. при тестировании)
Я попытался выяснить, как я могу зарегистрировать задачу динамически, например, из конструктора, но сельдерей разделяет код между рабочими, поэтому почему это кажется невозможным.

Спасибо за вашу помощь!

Ответ 1

Сельдерей имеет экспериментальную поддержку использования методов в качестве задач с версии 3.0.

Документация для этого находится в celery.contrib.methods, а также упоминаются некоторые предостережения, о которых вы должны знать:

http://docs.celeryproject.org/en/latest/reference/celery.contrib.methods.html

Знать: поддержка contrib.methods удалена из Celery с 4.0

Ответ 2

Если у вас есть:

    a = A()

вы можете сделать:

    A.foo.delay(a, param0, .., paramN)

Приветствия

Ответ 3

У Джереми Саттерфилда есть чистое и прямое руководство для написания задач на основе классов, если это то, что вы хотите выполнить. Вы можете проверить здесь.

Магия в основном расширяет класс celery.Task, включая метод run(), как-то вроде этого:

from celery import Task

class CustomTask(Task):
    ignore_result = True

    def __init__(self, arg):
        self.arg = arg

    def run(self):
        do_something_with_arg(self.arg)

а затем выполните задачу следующим образом:

your_arg = 3

custom_task = CustomTask()
custom_task.delay(your_arg)

Я не уверен, нужна ли часть ignore_result = True или нет BTW.

Ответ 4

Для меня единственное, что работает, - celery.current_app, потому что только это передает метод self в метод.

Итак, это должно выглядеть так:

from celery import current_app
from celery.contrib.methods import task_method

class A:
@current_app.task(filter=task_method, name='A.foo')
def foo(self, bar):
    ...

Имя должно использоваться, если у вас есть метод с тем же именем в разных классах.