How to wait for asynchronous subtasks to complete in Celery

user_14206864 0
Viewed 848

Here are the tasks:

from celery import shared_task
from celery import Celery
import time

celery = Celery("example_tasks", backend="rpc://", broker="amqp://")


@shared_task
def parent_task():
    print("Starting parent task.")
    for i in range(2):
        result = child_task.delay(i)


@shared_task
def child_task(index):
    time.sleep(5)
    print(f"Starting child task {index}.")

Here is the main program:

from celery import Celery
import time
from example_tasks import parent_task

app = Celery("tasks", backend="rpc://", broker="amqp://")

result = parent_task.delay()

At the end of the main program, what is the best way to wait until all tasks (subtasks included) have finished?

celery python
1 Answers

TL;DR

Use list(result.collect()).

collect(intermediate=False, **kwargs)

Collect results as they return.

Iterator, like get() will wait for the task to complete, but will also follow AsyncResult and ResultSet returned by the task, yielding (result, value) tuples for each result in the tree.

see docs for more details.

Explaination

The AsyncResult.collect() method is described in page Canvas: Designing Work-flows and looks like that this method only works for workflows, you need some primitives to build a result graph, but actually Celery will automaticly find the task parent when task was created(by self.current_worker_task).

That is to say, when you create a subtask, you build a result graph and AsyncResult.collect() is for you.

user_43316120
Related