How to create several chains in a for loop with Celery and group them all?

user_3423825 0
Viewed 337

I'm trying to create several chains by iterating through a list, and group them, but Celery complains about task2 saying that too many arguments are given. How could I achieve that?

This is how my code looks like:

@shared_task(bind=True)
def update(self):

    # list of chains
    ch = [chain(task1.s(exid),
                task2.s(exid),
                task3.s(exid),
                task4.s(exid)
                )()
          for exid in list_exid]

    gp = group(*ch)
    gp.delay()


@shared_task(bind=True)
def task1(self, exid):
    do_stuff()

@shared_task(bind=True)
def task2(self, exid):
    do_stuff()

@shared_task(bind=True)
def task3(self, exid):
    do_stuff()

@shared_task(bind=True)
def task4(self, exid):
    do_stuff()

And the error:

TypeError: task2() takes 1 positional argument but 3 were given
celery python
1 Answers

I answer my own question because I have found the solution to my problem. It was caused by using the wrong s() signature.

It is better like this:

@shared_task(bind=True)
def update(self):

    # list of chains
    ch = [chain(task1.si(exid),
                task2.si(exid),
                task3.si(exid),
                task4.si(exid)
                )()
          for exid in list_exid]

    gp = group(*ch)
    gp.delay()

EDIT

As stated in the documentation the use of .si creates immutable signatures meaning returned value of the previous task is ignored. I'm not sure what is the implication when a task from the chain fails

user_34238250
Related