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

Code before:
                tasks = [t.clone() for t in self.tasks]
                link_error = self.options.get('link_error', [])
                for sig in link_error:
                    for task in tasks:
                        task.on_error(sig)
                return _chain(seq_concat_item(tasks, other), app=self._app)
            # task | group() -> chain
            return _chain(self, other, app=self.app)

        if not isinstance(self, _chain) and isinstance(other, _chain):
            # task | chain -> chain
            tasks = [t.clone() for t in other.tasks]
            link_error = other.options.get('link_error', [])
            for sig in link_error:
                for task in tasks:
                    task.on_error(sig)
            return _chain(seq_concat_seq((self,), tasks), app=self._app)
        elif isinstance(other, _chain):
            # chain | chain -> chain
            # assign chain's link_error sugnatures to each chain's task
            tasks = [t.clone() for t in self.tasks]
            link_error = self.options.get('link_error', [])
            for sig in link_error:
                for task in tasks:
                    task.on_error(sig)
            other_tasks = [t.clone() for t in other.tasks]
            link_error = other.options.get('link_error', [])
            for sig in link_error:
                for task in other_tasks:
                    task.on_error(sig)
            return _chain(seq_concat_seq(tasks, other_tasks), app=self._app)
        elif isinstance(self, chord):
            # chord(ONE, body) | other -> ONE | body | other
            # chord with one header task is unecessary.
            if len(self.tasks) == 1:
                return self.tasks[0] | self.body | other
            # chord | task ->  attach to body
            sig = self.clone()
            sig.body = sig.body | other
            return sig
        elif isinstance(other, Signature):
            if isinstance(self, _chain):
                if isinstance(self.tasks[-1], group):
                    # CHAIN [last item is group] | TASK -> chord
                    sig = self.clone()
                    sig.tasks[-1] = chord(
                        sig.tasks[-1], other, app=self._app)
                    return sig
                elif isinstance(self.tasks[-1], chord):
                    # CHAIN [last item is chord] -> chain with chord body.
                    sig = self.clone()

Review comment (refactor):
Seems this is a repeated pattern, the more DRY approach would be to use a function for these 5 occurrences.

Code after:
                return _chain(seq_concat_item(
                    self.unchain_tasks(), other), app=self._app)
            # task | group() -> chain
            return _chain(self, other, app=self.app)

        if not isinstance(self, _chain) and isinstance(other, _chain):
            # task | chain -> chain
            return _chain(seq_concat_seq(
                (self,), other.unchain_tasks()), app=self._app)
        elif isinstance(other, _chain):
            # chain | chain -> chain
            return _chain(seq_concat_seq(
                self.unchain_tasks(), other.unchain_tasks()), app=self._app)
        elif isinstance(self, chord):
            # chord(ONE, body) | other -> ONE | body | other
            # chord with one header task is unecessary.
            if len(self.tasks) == 1:
                return self.tasks[0] | self.body | other
            # chord | task ->  attach to body
            sig = self.clone()
            sig.body = sig.body | other
            return sig
        elif isinstance(other, Signature):
            if isinstance(self, _chain):
                if isinstance(self.tasks[-1], group):
                    # CHAIN [last item is group] | TASK -> chord
                    sig = self.clone()
                    sig.tasks[-1] = chord(
                        sig.tasks[-1], other, app=self._app)
                    return sig
                elif isinstance(self.tasks[-1], chord):
                    # CHAIN [last item is chord] -> chain with chord body.
                    sig = self.clone()
                    sig.tasks[-1].body = sig.tasks[-1].body | other
                    return sig
                else:
                    # chain | task -> chain
                    return _chain(seq_concat_item(
                        self.unchain_tasks(), other), app=self._app)
            # task | task -> chain
            return _chain(self, other, app=self._app)
        return NotImplemented

    def election(self):
        type = self.type
        app = type.app
        tid = self.options.get('task_id') or uuid()

        with app.producer_or_acquire(None) as P:
            props = type.backend.on_task_call(P, tid)
            app.control.election(tid, 'task', self.clone(task_id=tid, **props),