This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""
        Don't crash when loading an invalid (contains a cycle) DAG file.
        Don't load the dag into the DagBag either
        """
        
        def basic_cycle():
            import datetime
            from airflow.models import DAG
            from airflow.operators.empty import EmptyOperator
            dag_name = "cycle_dag"
            default_args = {"owner": "owner1", "start_date": datetime.datetime(2016, 1, 1)}
            dag = DAG(dag_name, default_args=default_args)
            
            with dag:
                op_a = EmptyOperator(task_id="A")
                op_a.set_downstream(op_a)
            return dag
        test_dag = basic_cycle()
        
        assert len(test_dag.subdags) == 0
        
        dagbag, found_dags, file_path = self.process_dag(basic_cycle)
        
        
        self.validate_dags(test_dag, found_dags, dagbag, should_be_found=False)
        assert file_path in dagbag.import_errors
        
        def nested_subdag_cycle():