Complex workflow on Celery
<p>I am trying to build a workflow basing on Celery. I use groups and chords.</p>

<p>In the example below there are independent groups ([mytask1, mytask1, mytask1, ..] -> myfinaltask1) where <code>mytask1</code> might be executed in parallel, but <code>myfinaltask1</code> should be called after each group.</p>

<p>Code:</p>

<pre><code>def func1(date):
    subtasks = []
    for filepath in all_files:
        kwargs = {'date': date, 'hfile': filepath}
        subtask = mytask1.subtask(kwargs=kwargs)
        subtasks.append(subtask)

    chrd = chord(subtasks)
    chrdr = chrd(myfinaltask1.s(kwargs={'date': date}))
    return chrdr


def main(all_dates):
    subtasks = []
    for ad in all_dates:
        subtasks.append(func1(ad))

    g = group(subtasks)
    gr = g.apply_async()
    results = gr.get(propagate=False)  # sync wait!


main([2014, 2015, 2016])
</code></pre>

<p>Exception thrown:</p>

<pre><code>File "/mypath/get_evi.py", line 265, in get_evi_year
    gr = g.apply_async()
File "/opt/venv/lib/python3.5/site-packages/celery/canvas.py", line 502, in apply_async
    type = self.type
File "/opt/venv/lib/python3.5/site-packages/celery/canvas.py", line 569, in type
    return self.app.tasks[self['task']]
File "/opt/venv/lib/python3.5/site-packages/celery/canvas.py", line 560, in app
    return self._app or (self.tasks[0].app if self.tasks else current_app)
AttributeError: 'bool' object has no attribute 'app'
</code></pre>

<p>What do I do wrong?</p>

<p>It seems you've forgot to wrap <code>subtasks</code> to <code>group</code>.</p>

<pre><code>def func1(date):
    subtasks = []
    for filepath in all_files:
        kwargs = {'date': date, 'hfile': filepath}
        subtask = mytask1.subtask(kwargs=kwargs)
        subtasks.append(subtask)

    chrd = chord(header=group(subtasks), body=myfinaltask1.subtask(kwargs={'date': date}))
    return chrdr
</code></pre>

python celery