Execute multiple celery tasks inside a celery task

Unfortunately, it doesn't work this way. You need to create a Celery Group and trigger the group execution. To be able to "tie" all the executions, pass variables to the tasks when adding them to the Group, so you can make all of them report to a central resource, using this reference. Then check the central resource to see if all tasks in the Group have already been reported.

I will paste here a code I've been using to manage Celery Group tasks. You do need a central resource (like a database for example) and make your tasks report to it, so you can go there and check the progress. But the code I'm sharing can help you to understand how to manage a Celery Group. I use them so I do not handle Celery technicalities when coding.

The Django dependencies you can remove and swap for your project needs.

import importlib

from celery import group
from celery.result import AsyncResult

from django.core.exceptions import ValidationError
from django.conf import settings


class CeleryEntity:

    def __init__(self, uuid=None):
        self.uuid = uuid
        self._celery_app = None
        self._result = None
        self._initialize()

    def __str__(self):
        return self.uuid

    def _initialize(self, celery_app=None):
        if not celery_app:
            celery_app_reference = getattr(settings, 'CELERY_TOOLS_APP', None)
            celery_app = getattr(importlib.import_module(celery_app_reference[0]), celery_app_reference[1])
        self._celery_app = celery_app

    def _validate_celery_app(self):
        self._initialize()
        if not self._celery_app:
            raise ValidationError(
                'No Celery App Object. Check your CELERY_TOOLS_APP on settings.py')

  

class CeleryTaskGroup(CeleryEntity):

    def __init__(self, uuid=None, save=False):
        super().__init__(uuid)
        self.save = save
        self._tasks = []
        self._result_set = None

    def __str__(self):
        return self.uuid

    def _initialize(self, celery_app=None):
        if not celery_app:
            celery_app_reference = getattr(settings, 'CELERY_TOOLS_APP', None)
            celery_app = getattr(importlib.import_module(celery_app_reference[0]), celery_app_reference[1])
        self._celery_app = celery_app

    def _validate_tasks(self):
        if not self._tasks:
            return ValidationError('No Tasks defined. Add tasks using append_task method')

    def _get_result_set(self):
        if not self._result_set:
            self._validate_celery_app()
            results = self._celery_app.backend.restore_group(str(self.uuid))
            self._result_set = results
        return self._result_set

    def length(self):
        return len(self._tasks)

    def stats(self):
        """ Process statistics dictionary """

        results = self._get_result_set()

        if results:
            ready = results.ready()
            successful = results.successful()
            completed = results.completed_count()
        else:
            ready = False
            successful = False
            completed = 0

        stats = {
            'ready': ready,
            'successful': successful,
            'completed': completed,
        }

        return stats

    def append_task(self, task_name, params=()):
        self._tasks.append(self._celery_app.signature(task_name, params))

    def run_now(self, queue=None, wait=False):

        self._validate_celery_app()
        self._validate_tasks()

        job = group(self._tasks)
        group_results = job.apply_async(queue=queue)
        if self.save:
            group_results.save()

        self.uuid = group_results.id

        if wait:
            group_results.join()

        return group_results

    def revoke_process(self):
        grp = self._get_group()
        if grp:
            grp.revoke(terminate=True)
        self._delete_group()

    def _get_group(self):
        self._validate_celery_app()
        grp = self._celery_app.backend.restore_group(str(self.uuid))
        return self._celery_app.GroupResult(id=str(self.uuid), results=grp) if grp else None

    def _delete_group(self):
        self._validate_celery_app()
        self._celery_app.backend.delete_group(str(self.uuid))


class CeleryTask(CeleryEntity):

    def _get_result(self):
        if not self._result:
            self._validate_celery_app()
            self._result = AsyncResult(str(self.uuid), app=self._celery_app)
        return self._result

    def status(self):
        result = self._get_result()
        workers = self._celery_app.control.inspect().registered()
        if result.id == 'None' or not workers:
            return 'NO_TASK'
        return result.state

    def revoke(self):
        result = self._get_result()
        result.revoke(terminate=True)

    def run(self, task_name, params, queue=None):
        result = self._celery_app.signature(task_name, params).apply_async(queue=queue)
        self.uuid = result.id
        return result