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_0 = DAG("parent.op_subdag_0", default_args=default_args)
                    EmptyOperator(task_id="subdag_0.task", dag=subdag_0)
                    return subdag_0
                def subdag_1():
                    subdag_1 = DAG("parent.op_subdag_1", default_args=default_args)
                    EmptyOperator(task_id="subdag_1.task", dag=subdag_1)
                    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 = standard_subdag()
        
        assert len(test_dag.subdags) == 2
        
        dagbag, found_dags, _ = self.process_dag(standard_subdag)
        
        
        self.validate_dags(test_dag, found_dags, dagbag)
        
        def nested_subdags():