I am successfully using chord in my system. But, Now I have a case that, I have to run numbers of chord sequentially (second one starts when first ends). So, I was trying to use chain of chords. But it doesn't work for me.
@app.task(bind=True)
def some_celery_beat_worker(self):
feed_chain = []
for feed in feed_list:
celery_task_list = [perform single_task.si(url)
for url in some_url_list]
per_feed_chord = chord(celery_task_list, chord_callback.si(feed['_id'], feed['xml_file_name']))
feed_chain.append(per_feed_chord)
chain(*feed_chain).delay()
Getting this as traceback:
Traceback (most recent call last): File "python3.4/site-packages/celery/app/trace.py", line 374, in trace_task R = retval = fun(*args, **kwargs) File "python3.4/site-packages/celery/app/trace.py", line 629, in __protected_call__ return self.run(*args, **kwargs) File "workers.py", line 156, in joblist_updater_worker chain(*feed_chain).delay() File "python3.4/site-packages/celery/canvas.py", line 182, in delay return self.apply_async(partial_args, partial_kwargs) File "/python3.4/site-packages/celery/canvas.py", line 566, in apply_async dict(self.options, **options) if options else self.options)) File "python3.4/site-packages/celery/canvas.py", line 596, in run first_task.apply_async(**options) File "python3.4/site-packages/celery/canvas.py", line 1241, in apply_async return (self.tasks[0] | body).set(task_id=task_id).apply_async( KeyError: 0
What I need is to fix this particular workflow, or any alternative workflow, which can solve this particular problem (running numbers of chord one after another)