Celery chain task on group

Viewed 282
@shared_task
def process_record(x):
    return [1,2,4,4,5,6]

@shared_task
def add(pro_id):
    return pro_id + 10


@shared_task
def dmap(it, callback):
    callback = subtask(callback)
    group_args = [callback.clone([arg,]) for arg in it]
    return group(group_args)()

ch = (process_record.s(10) | dmap.s(add.s()))
ch.apply_async()

the above code works fine, for the above two tasks. Now, I need to add more task which is to be called after add (with its return value) task.

@shared_task
def mul(add_id):
    return add_id * 2

I have tired ch = (process_record.s(10) | dmap.s(add.s()) | mul.s()) it does not work since return type from add is GroupResult.

How can mul task can be chained at end ?

resource: How to chain a Celery task that returns a list into a group?

0 Answers
Related