airflow/models/dagbag.py
def bag_dag(self, dag, parent_dag, root_dag):
        """
        Adds the DAG into the bag, recurses into sub dags.
        Throws AirflowDagCycleException if a cycle is detected in this dag or its subdags
        """

        dag.test_cycle()  # throws if a task cycle is found

        dag.resolve_template_files()
        dag.last_loaded = timezone.utcnow()

        for task in dag.tasks:
            settings.policy(task)

        subdags = dag.subdags

        try:
            for subdag in subdags:
                subdag.full_filepath = dag.full_filepath
                subdag.parent_dag = dag
                subdag.is_subdag = True
                self.bag_dag(subdag, parent_dag=dag, root_dag=root_dag)

            self.dags[dag.dag_id] = dag
            self.log.debug('Loaded DAG %s', dag)
        except AirflowDagCycleException as cycle_exception:
            # There was an error in bagging the dag. Remove it from the list of dags
            self.log.exception('Exception bagging dag: %s', dag.dag_id)
            # Only necessary at the root level since DAG.subdags automatically
            # performs DFS to search through all subdags
            if dag == root_dag:
                for subdag in subdags:
                    if subdag.dag_id in self.dags:
                        del self.dags[subdag.dag_id]
            raise cycle_exception
def bag_dag(self, dag, parent_dag, root_dag):
        dag.test_cycle()  # throws if a task cycle is found

        dag.resolve_template_files()
        dag.last_loaded = timezone.utcnow()

        for task in dag.tasks:
            settings.policy(task)

        subdags = dag.subdags

        try:
            for subdag in subdags:
                subdag.full_filepath = dag.full_filepath
                subdag.parent_dag = dag
                subdag.is_subdag = True
                self.bag_dag(subdag, parent_dag=dag, root_dag=root_dag)

            self.dags[dag.dag_id] = dag
            self.log.debug('Loaded DAG %s', dag)
        except AirflowDagCycleException as cycle_exception:
            # There was an error in bagging the dag. Remove it from the list of dags
            self.log.exception('Exception bagging dag: %s', dag.dag_id)
            # Only necessary at the root level since DAG.subdags automatically
            # performs DFS to search through all subdags
            if dag == root_dag:
                for subdag in subdags:
                    if subdag.dag_id in self.dags:
                        del self.dags[subdag.dag_id]
            raise cycle_exception
def bag_dag ( self , dag , parent_dag , root_dag ) : dag . test_cycle ( ) # throws if a task cycle is found dag . resolve_template_files ( ) dag . last_loaded = timezone . utcnow ( ) for task in dag . tasks : settings . policy ( task ) subdags = dag . subdags try : for subdag in subdags : subdag . full_filepath = dag . full_filepath subdag . parent_dag = dag subdag . is_subdag = True self . bag_dag ( subdag , parent_dag = dag , root_dag = root_dag ) self . dags [ dag . dag_id ] = dag self . log . debug ( 'Loaded DAG %s' , dag ) except AirflowDagCycleException as cycle_exception : # There was an error in bagging the dag. Remove it from the list of dags self . log . exception ( 'Exception bagging dag: %s' , dag . dag_id ) # Only necessary at the root level since DAG.subdags automatically # performs DFS to search through all subdags if dag == root_dag : for subdag in subdags : if subdag . dag_id in self . dags : del self . dags [ subdag . dag_id ] raise cycle_exception
Adds the DAG into the bag, recurses into sub dags.
        Throws AirflowDagCycleException if a cycle is detected in this dag or its subdags
Adds the DAG into the bag recurses into sub dags . Throws AirflowDagCycleException if a cycle is detected in this dag or its subdags