I need to make requests to an API with some params and process data from requests one by one. But sometime response data is paginated, means I need to make extra request with same params.
Celery group allow us to run tasks one by one, but if tasks produce child tasks, and child tasks can produce more child tasks...,
Is there a way to wait all children tasks finished before run next task in the group? Or maybe celery give us better approach to solve my task?
def some_api_call(start_date=None, end_date=None, token_value=None):
pass
@celery_app.task(name='run_task', max_retries=None, bind=True)
def run_task(self):
group_items = [
task1.s('2020-01-02', '2020-01-03'),
task1.s('2020-01-05', '2020-01-01'),
task1.s('2020-01-010', '2020-01-04'),
]
group(group_items)()
@celery_app.task(name='task1', max_retries=None, bind=True)
def task1(self, start_date=None, end_date=None, token_value=None, *args, **kwargs):
res = some_api_call(start_date, end_date, token_value)
if res['token_value']:
# NEXT ELEMENT IN THE GROUP SHOULD WAIT UNTIL NESTED CHILD TASKS DONE
task1.delay(token_value=token_value)
Broker - Redis.
My possible solutions[pseudocode]:
- Wait until child tasks done in a parent task
res = task1.delay(token_value=token_value)
res.get()
Solution is bad - we block tread. Not sure if celery has await alternative.
- User
task.retry()to check if child task is finished.
taskid = task1.delay(token_value=token_value)
if AsyncResult(taskid).state != "successes":
self.retry()
So we will retry parent tasks until children tasks finished and don't block the thread. But like in the first solution: parent tasks processed its data, but state would be retry.
