def topological_sort(self):
<EOL>graph_unsorted = OrderedDict((task.task_id, task) for task in self.tasks)<EOL>graph_sorted = []<EOL>if len(self.tasks) == <NUM_LIT:0>:<EOL><INDENT>return tuple(graph_sorted)<EOL><DEDENT>while graph_unsorted:<EOL><INDENT>acyclic = False<EOL>for node in list(graph_unsorted.values()):<EOL><INDENT>for edge in node.upstream_list:<EOL><INDENT>if edge.task_id in graph_unsorted:<EOL><INDENT>break<EOL><DEDENT><DEDENT>else:<EOL><INDENT>acyclic = True<EOL>del graph_unsorted[node.task_id]<EOL>graph_sorted.append(node)<EOL><DEDENT><DEDENT>if not acyclic:<EOL><INDENT>raise AirflowException("<STR_LIT>"<EOL>.format(self.dag_id))<EOL><DEDENT><DEDENT>return tuple(graph_sorted)<EOL>
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/

:return: list of tasks in topological order