python - Celery 任务链和访问 **kwargs

Question

我的情况与概述的情况类似 here ，除了不是用多个参数链接任务，我想链接返回包含多个条目的字典的任务。

这是——非常松散和抽象的——我正在尝试做的事情:

任务.py

@task()
def task1(item1=None, item2=None):
  item3 = #do some stuff with item1 and item2 to yield item3
  return_object = dict(item1=item1, item2=item2, item3=item3)
  return return_object

def task2(item1=None, item2=None, item3=None):
  item4 = #do something with item1, item2, item3 to yield item4
  return_object = dict(item1=item1, item2=item2, item3=item3, item4=item4)
  return return_object

在 ipython 中工作，我能够单独和异步地调用 task1，没有任何问题。

我还可以使用 task1 返回的结果作为双星参数单独调用 task2:

>>res1 = task1.s(item1=something, item2=something_else).apply_async()
>>res1.status
'SUCCESS'
>>res2 = task2.s(**res1.result).apply_async()
>>res2.status
'SUCCESS

但是，我最终想要实现的是与上面相同的最终结果，但是通过一个链，在这里，我无法弄清楚如何不使用(位置)参数来实例化 task2由 task1 返回，但 task1.result 为 **kwargs:

chain_result = (task1.s(item1=something, item2=something_else) | task2.s()).apply_async()  #THIS DOESN'T WORK!

我怀疑我可以返回并重写我的任务，以便它们返回位置参数而不是字典，这可能会解决问题，但在我看来应该有某种方法可以访问任务 2 中任务 1 的返回对象具有**双星的等效功能。我还怀疑我在这里遗漏了一些关于 Celery 子任务实现或 *args 与 **kwargs 的相当明显的东西。

希望这是有道理的。并提前感谢您提供任何提示。

Answer 1

这是我对这个问题的看法，使用抽象任务类:

from __future__ import absolute_import
from celery import Task
from myapp.tasks.celery import app   


class ChainedTask(Task):
    abstract = True    

    def __call__(self, *args, **kwargs):
        if len(args) == 1 and isinstance(args[0], dict):
            kwargs.update(args[0])
            args = ()
        return super(ChainedTask, self).__call__(*args, **kwargs)

@app.task(base=ChainedTask)
def task1(x, y):
    return {'x': x * 2, 'y': y * 2, 'z': x * y}    


@app.task(base=ChainedTask)
def task2(x, y, z):
    return {'x': x * 3, 'y': y * 3, 'z': z * 2}

您现在可以这样定义和执行您的链:

from celery import chain

pipe = chain(task1.s(x=1, y=2) | task2.s())
pipe.apply_async()