Function replace_with_stamped_task

t/integration/tasks.py:519–522

(self: StampedTaskOnReplace, replace_with=None)

    @shared_task(bind=True, base=StampedTaskOnReplace)
    def replace_with_stamped_task(self: StampedTaskOnReplace, replace_with=None):
        if replace_with is None:
            replace_with = replaced_with_me.s()
        self.replace(signature(replace_with))


@shared_task(bind=True, acks_late=True)