Celery how to pass arguments to linked task (which is bound)?

Viewed 613

I have 2 tasks, which one after success should start second and pass results. Both task are bind=True because of logic of reporting retries.

I call this task by:

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

But have no idea how to pass result to bar task. Docs show only example without bind, and in my case result is not passed to second task bar

Traceback show that y is not passed:

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}
1 Answers

I found out that I can just use:

task = (foo.s(x) | bar.s()).apply_async() and it's working as intended.

Related