title: Collect results from parallel Celery task executions
body: <p>I am trying to achieve a rather simple Celery workflow, where I receive the result of multiple parallel invocations of the same task as a tuple (or list).</p>
  
  <pre><code>@app.task
  def add(x, y):
      return x + y
  
  @app.task
  def master():
      return group(add.s(1, 2), add.s(3, 4))()
  </code></pre>
  
  <p>From this, I would like to retrieve <code>(3, 7)</code> <em>in a generic way</em>, that is, in a way that does not depend on the workflow itself. I am looking for some sort of "reduce async result graph to primitives" operation. I have experimented with the following (I have replaced result IDs with <code>#num</code> for brevity)</p>
  
  <pre><code>r = master.delay()
  r.get()      # &lt;GroupResult: #1 [#2, #3]&gt;
  r.collect()  # [(&lt;AsyncResult: #0&gt;, &lt;GroupResult #1 [#2, #3]&gt;),
               #  (&lt;GroupResult: #1 [#2, #3]&gt;, [3, 7])
               #  (&lt;AsyncResult: #2&gt;, 3),
               #  (&lt;GroupResult: #3&gt;, 7)]
  </code></pre>
  
  <p><code>r.get()</code> returns a wrapper around two <code>AsyncResult</code> IDs, so I would have to process each one recursively. <code>r.collect()</code> is close, but it recurses too deeply.</p>
  
  <p>I could do something like</p>
  
  <pre><code>r.children[0].get()
  </code></pre>
  
  <p>but this is not generic, since it explicitly depends on the structure of the result graph. Also, I could iterate through <code>r.collect()</code> until I find a tuple whose value is not an instance of <code>ResultBase</code>, like</p>
  
  <pre><code>next(value for _, value in r.collect() if not isinstance(value, ResultBase))
  </code></pre>
  
  <p>but I am not sure if this is actually correct in all cases, and I was hoping that there is a more elegant way to do this.</p>
  
  <p>If there is a way of restructuring the <code>master</code> task to make retrieving the results easier, I am open to it, just as long as the subtasks are launched in parallel. Any suggestions would be appreciated. Thank you in advance.</p>
  
  <hr>
  
  <p><strong>EDIT</strong> A related issue is that if I want to retrieve the task results in a non-blocking way (for example, by manually polling <code>r.status</code> before calling <code>r.get()</code> or <code>r.collect()</code>, I cannot simply do this</p>
  
  <pre><code>r = master.delay()
  
  # some time later...
  if r.status in READY_STATES:
      r.get()
  </code></pre>
  
  <p>because <code>r</code> is an <code>AsyncResult</code> which resolves to a <code>GroupResult</code>, i.e. it completes before the <code>GroupResult</code> or its children. Is there a way to invoke the group in a way that "skips" the top-level <code>AsyncResult</code>? This would solve both issues, since <code>r.status</code> and <code>r.get()</code> would reflect the status and value of the child task, respectively.</p>