This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
subdag_d = DAG("parent.op_subdag_1.opSubdag_D", default_args=default_args)
                    EmptyOperator(task_id="subdag_d.task", dag=subdag_d)
                    return subdag_d
                def subdag_0():
                    subdag_0 = DAG("parent.op_subdag_0", default_args=default_args)
                    SubDagOperator(task_id="opSubdag_A", dag=subdag_0, subdag=subdag_a())
                    SubDagOperator(task_id="opSubdag_B", dag=subdag_0, subdag=subdag_b())
                    return subdag_0
                def subdag_1():
                    subdag_1 = DAG("parent.op_subdag_1", default_args=default_args)
                    SubDagOperator(task_id="opSubdag_C", dag=subdag_1, subdag=subdag_c())
                    SubDagOperator(task_id="opSubdag_D", dag=subdag_1, subdag=subdag_d())
                    return subdag_1
                op_subdag_0 = SubDagOperator(task_id="op_subdag_0", dag=dag, subdag=subdag_0())
                op_subdag_1 = SubDagOperator(task_id="op_subdag_1", dag=dag, subdag=subdag_1())
                op_a = EmptyOperator(task_id="A")
                op_a.set_downstream(op_subdag_0)
                op_a.set_downstream(op_subdag_1)
            return dag
        test_dag = nested_subdags()
        
        assert len(test_dag.subdags) == 6
        
        dagbag, found_dags, filename = self.process_dag(nested_subdags)
        
        
        self.validate_dags(test_dag, found_dags, dagbag)
        for dag in dagbag.dags.values():
            assert dag.fileloc == filename
    def test_skip_cycle_dags(self):