'When invoked, this routine will raise an exception if a cycle is
detected downstream from self. It is invoked when tasks are added to
the DAG to detect cycles.'

def detect_downstream_cycle(self, task=None):

    if (not task):
        task = self
    for t in self.get_direct_relatives():
        if (task is t):
            msg = u'Cycle    detected    in    DAG.    Faulty    task:    {0}'.format(task)
            raise AirflowException(msg)
        else:
            t.detect_downstream_cycle(task=task)
    return False
