Repository: celery/celery
File: celery/canvas.py (Python)

Code before:
    called.  A chord is essentially a callback for a group of tasks.

    The body is applied with the return values of all the header
    tasks as a list.

    Example:

        The chord:

        .. code-block:: pycon

            >>> res = chord([add.s(2, 2), add.s(4, 4)])(sum_task.s())

        is effectively :math:`\Sigma ((2 + 2) + (4 + 4))`:

        .. code-block:: pycon

            >>> res.get()
            12
    """

    @classmethod
    def from_dict(cls, d, app=None):
        options = d.copy()
        args, options['kwargs'] = cls._unpack_args(**options['kwargs'])
        args = [maybe_signature(task) for task in args]
        return _upgrade(d, cls(*args, app=app, **options))

    @staticmethod
    def _unpack_args(header=None, body=None, **kwargs):
        # Python signatures are better at extracting keys from dicts
        # than manually popping things off.
        return (header, body), kwargs

    def __init__(self, header, body=None, task='celery.chord',
                 args=None, kwargs=None, app=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        Signature.__init__(
            self, task, args,
            {'kwargs': kwargs, 'header': _maybe_group(header, app),
             'body': maybe_signature(body, app=app)}, app=app, **options
        )
        self.subtask_type = 'chord'

    def __call__(self, body=None, **options):
        return self.apply_async((), {'body': body} if body else {}, **options)

    def freeze(self, _id=None, group_id=None, chord=None,
               root_id=None, parent_id=None, group_index=None):
        # pylint: disable=redefined-outer-name

Review comment (bug):
My more complex failure case hits this code somewhere 9asserting false makes it die) but this change doesn't seem to be required to fix my breakage (the change to groups fixes my breakage). In any case, this seems like a legit change to make.

Code after:
                                  parent_id=parent_id,
                                  group_index=group_index)
                group_index += 1

    def __repr__(self):
        if self.tasks:
            return remove_repeating_from_task(
                self.tasks[0]['task'],
                f'group({self.tasks!r})')
        return 'group(<empty>)'

    def __len__(self):
        return len(self.tasks)

    @property
    def app(self):
        app = self._app
        if app is None:
            try:
                app = self.tasks[0].app
            except LookupError:
                pass
        return app if app is not None else current_app


@Signature.register_type()
class chord(Signature):
    r"""Barrier synchronization primitive.

    A chord consists of a header and a body.

    The header is a group of tasks that must complete before the callback is
    called.  A chord is essentially a callback for a group of tasks.

    The body is applied with the return values of all the header
    tasks as a list.

    Example:

        The chord:

        .. code-block:: pycon

            >>> res = chord([add.s(2, 2), add.s(4, 4)])(sum_task.s())

        is effectively :math:`\Sigma ((2 + 2) + (4 + 4))`:

        .. code-block:: pycon

            >>> res.get()
            12