python - Celery:如何为基于类的任务创建签名？

Question

Python 3.7、 celery 4.3

我注册了一个基于类的任务:

class TaskA(celery_app.Task):
    name = 'task_a'

    def run(self, my_param):
       pass

celery_app.tasks.register(TaskA())

我想创建一个任务链，其中 TaskA 应该是其中一个链接。

我是这样做的:

my_param = "some value"

result = celery.chain(
    TaskA.s(my_param),
    # ...
)()

我收到错误:

TypeError: s() missing 1 required positional argument: 'self'

Answer 1

替换代码:

my_param = "some value"

result = celery.chain(
    TaskA.s(my_param),
    # ...
)()

与:

my_param = "some value"

task_ = TaskA() # <------
result = celery.chain(
    task_.s(my_param),
    # ...
)()