Celery如何将参数传递给链接任务（这是绑定的）？

Question

我有两个任务，其中一个任务成功后应该开始第二个任务并通过结果。两个任务都是 bind=True 因为报告重试的逻辑。

我通过以下方式调用此任务：

async def foobar(x: int):
    task = foo.apply_async(kwargs={'x': x}, link=bar.si())

但不知道如何将结果传递给栏任务。 Docs 仅显示没有绑定的示例，在我的情况下结果不是传递给第二个任务 bar

回溯显示 y 未传递：

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/celery/app/trace.py", line 503, in trace_task
    signature(callbacks[0], app=app).apply_async(
  File "/usr/local/lib/python3.8/site-packages/celery/canvas.py", line 219, in apply_async
    return _apply(args, kwargs, **options)
  File "/usr/local/lib/python3.8/site-packages/celery/app/task.py", line 537, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: bar() missing 1 required positional argument: 'y'

@celery_app.task(name="foo", bind=True,
                 autoretry_for=(Exception,),
                 default_retry_delay=1,
                 retry_backoff=2,  # <- wait 1s before the 1st retry
                 max_retries=2,
                 retry_jitter=False)
def foo(self, x: int, *args):
    logger.info(f'{self.request.id}, countdown: {self.request.retries}, max_retries: {self.max_retries}')
    return {"x": x}


@celery_app.task(name="bar", bind=True,
                 autoretry_for=(Exception,),
                 default_retry_delay=1,
                 retry_backoff=2,  # <- wait 1s before the 1st retry
                 max_retries=2,
                 retry_jitter=False)
def bar(self, y: int):
    logger.info(f'{self.request.id}, countdown: {self.request.retries}, max_retries: {self.max_retries}')
    return {"y_which_should_be_x": y}

Answer 1

我发现我可以使用：

task = (foo.s(x) | bar.s()).apply_async() 并且它按预期工作。