Method replace

celery/app/task.py:949–1015

Replace this task, with a new task inheriting the task id. Execution of the host task ends immediately and no subsequent statements will be run. .. versionadded:: 4.0 Arguments: sig (Signature): signature to replace with. visitor (StampingVi

(self, sig)

                uuid=req.id, retry=retry, retry_policy=retry_policy, **fields)

    def replace(self, sig):
        """Replace this task, with a new task inheriting the task id.

        Execution of the host task ends immediately and no subsequent statements
        will be run.

        .. versionadded:: 4.0

        Arguments:
            sig (Signature): signature to replace with.
            visitor (StampingVisitor): Visitor API object.

        Raises:
            ~@Ignore: This is always raised when called in asynchronous context.
            It is best to always use ``return self.replace(...)`` to convey
            to the reader that the task won't continue after being replaced.
        """
        chord = self.request.chord
        if 'chord' in sig.options:
            raise ImproperlyConfigured(
                "A signature replacing a task must not be part of a chord"
            )
        if isinstance(sig, _chain) and not getattr(sig, "tasks", True):
            raise ImproperlyConfigured("Cannot replace with an empty chain")

        # Ensure callbacks or errbacks from the replaced signature are retained
        if isinstance(sig, group):
            # Groups get uplifted to a chord so that we can link onto the body
            sig |= self.app.tasks['celery.accumulate'].s(index=0)
        for callback in maybe_list(self.request.callbacks) or []:
            sig.link(callback)
        for errback in maybe_list(self.request.errbacks) or []:
            sig.link_error(errback)
        # If the replacement signature is a chain, we need to push callbacks
        # down to the final task so they run at the right time even if we
        # proceed to link further tasks from the original request below
        if isinstance(sig, _chain) and "link" in sig.options:
            final_task_links = sig.tasks[-1].options.setdefault("link", [])
            final_task_links.extend(maybe_list(sig.options["link"]))
        # We need to freeze the replacement signature with the current task's
        # ID to ensure that we don't disassociate it from the existing task IDs
        # which would break previously constructed results objects.
        sig.freeze(self.request.id)
        # Ensure the important options from the original signature are retained
        replaced_task_nesting = self.request.get('replaced_task_nesting', 0) + 1
        sig.set(
            chord=chord,
            group_id=self.request.group,
            group_index=self.request.group_index,
            root_id=self.request.root_id,
            replaced_task_nesting=replaced_task_nesting
        )

        # If the replaced task is a chain, we want to set all of the chain tasks
        # with the same replaced_task_nesting value to mark their replacement nesting level
        if isinstance(sig, _chain):
            for chain_task in maybe_list(sig.tasks) or []:
                chain_task.set(replaced_task_nesting=replaced_task_nesting)