airflow/models/dag.py
def topological_sort(self, include_subdag_tasks: bool = False):
        """
        Sorts tasks in topographical order, such that a task comes after any of its
        upstream dependencies.

        Heavily inspired by:
        http://blog.jupo.org/2012/04/06/topological-sorting-acyclic-directed-graphs/

        :param include_subdag_tasks: whether to include tasks in subdags, default to False
        :return: list of tasks in topological order
        """
        from airflow.operators.subdag import SubDagOperator  # Avoid circular import

        # convert into an OrderedDict to speedup lookup while keeping order the same
        graph_unsorted = OrderedDict((task.task_id, task) for task in self.tasks)

        graph_sorted = []  # type: List[BaseOperator]

        # special case
        if len(self.tasks) == 0:
            return tuple(graph_sorted)

        # Run until the unsorted graph is empty.
        while graph_unsorted:
            # Go through each of the node/edges pairs in the unsorted
            # graph. If a set of edges doesn't contain any nodes that
            # haven't been resolved, that is, that are still in the
            # unsorted graph, remove the pair from the unsorted graph,
            # and append it to the sorted graph. Note here that by using
            # using the items() method for iterating, a copy of the
            # unsorted graph is used, allowing us to modify the unsorted
            # graph as we move through it. We also keep a flag for
            # checking that graph is acyclic, which is true if any
            # nodes are resolved during each pass through the graph. If
            # not, we need to exit as the graph therefore can't be
            # sorted.
            acyclic = False
            for node in list(graph_unsorted.values()):
                for edge in node.upstream_list:
                    if edge.task_id in graph_unsorted:
                        break
                # no edges in upstream tasks
                else:
                    acyclic = True
                    del graph_unsorted[node.task_id]
                    graph_sorted.append(node)
                    if include_subdag_tasks and isinstance(node, SubDagOperator):
                        graph_sorted.extend(node.subdag.topological_sort(include_subdag_tasks=True))

            if not acyclic:
                raise AirflowException(f"A cyclic dependency occurred in dag: {self.dag_id}")

        return tuple(graph_sorted)

        Sorts tasks in topographical order, such that a task comes after any of its
        upstream dependencies.

        Heavily inspired by:
        http://blog.jupo.org/2012/04/06/topological-sorting-acyclic-directed-graphs/

        :param include_subdag_tasks: whether to include tasks in subdags, default to False
        :return: list of tasks in topological order
        
Sorts tasks in topographical order, such that a task comes after any of its
upstream dependencies.


Sorts tasks in topographical order such that a task comes after any of its upstream dependencies .
def topological_sort(self, include_subdag_tasks: bool = False):
        from airflow.operators.subdag import SubDagOperator  
        graph_unsorted = OrderedDict((task.task_id, task) for task in self.tasks)
        graph_sorted = []  
        if len(self.tasks) == 0:
            return tuple(graph_sorted)
        while graph_unsorted:
            acyclic = False
            for node in list(graph_unsorted.values()):
                for edge in node.upstream_list:
                    if edge.task_id in graph_unsorted:
                        break
                else:
                    acyclic = True
                    del graph_unsorted[node.task_id]
                    graph_sorted.append(node)
                    if include_subdag_tasks and isinstance(node, SubDagOperator):
                        graph_sorted.extend(node.subdag.topological_sort(include_subdag_tasks=True))
            if not acyclic:
                raise AirflowException(f"A cyclic dependency occurred in dag: {self.dag_id}")
        return tuple(graph_sorted)
def topological_sort ( self , include_subdag_tasks : bool = False ) : from airflow . operators . subdag import SubDagOperator graph_unsorted = OrderedDict ( ( task . task_id , task ) for task in self . tasks ) graph_sorted = [ ] if len ( self . tasks ) == 0 : return tuple ( graph_sorted ) while graph_unsorted : acyclic = False for node in list ( graph_unsorted . values ( ) ) : for edge in node . upstream_list : if edge . task_id in graph_unsorted : break else : acyclic = True del graph_unsorted [ node . task_id ] graph_sorted . append ( node ) if include_subdag_tasks and isinstance ( node , SubDagOperator ) : graph_sorted . extend ( node . subdag . topological_sort ( include_subdag_tasks = True ) ) if not acyclic : raise AirflowException ( f"A cyclic dependency occurred in dag: {self.dag_id}" ) return tuple ( graph_sorted )
Sorts tasks in topographical order, such that a task comes after any of its
upstream dependencies.
Sorts tasks in topographical order such that a task comes after any of its upstream dependencies .
"""
        Sorts tasks in topographical order, such that a task comes after any of its
        upstream dependencies.

        Heavily inspired by:
        http://blog.jupo.org/2012/04/06/topological-sorting-acyclic-directed-graphs/

        :param include_subdag_tasks: whether to include tasks in subdags, default to False
        :return: list of tasks in topological order
        """ # Avoid circular import # convert into an OrderedDict to speedup lookup while keeping order the same # type: List[BaseOperator] # special case # Run until the unsorted graph is empty. # Go through each of the node/edges pairs in the unsorted # graph. If a set of edges doesn't contain any nodes that # haven't been resolved, that is, that are still in the # unsorted graph, remove the pair from the unsorted graph, # and append it to the sorted graph. Note here that by using # using the items() method for iterating, a copy of the # unsorted graph is used, allowing us to modify the unsorted # graph as we move through it. We also keep a flag for # checking that graph is acyclic, which is true if any # nodes are resolved during each pass through the graph. If # not, we need to exit as the graph therefore can't be # sorted. # no edges in upstream tasks
list of tasks in topological order
list of tasks in topological order
self
include_subdag_tasks
bool
whether to include tasks in subdags, default to False
whether to include tasks in subdags default to False