This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
from airflow.models.dag import DAG, DagContext  
        top_level_dags = {(o, m) for m in mods for o in m.__dict__.values() if isinstance(o, DAG)}
        top_level_dags.update(DagContext.autoregistered_dags)
        DagContext.current_autoregister_module_name = None
        DagContext.autoregistered_dags.clear()
        found_dags = []
        for (dag, mod) in top_level_dags:
            dag.fileloc = mod.__file__
            try:
                dag.validate()
                self.bag_dag(dag=dag, root_dag=dag)
            except Exception as e:
                self.log.exception("Failed to bag_dag: %s", dag.fileloc)
                self.import_errors[dag.fileloc] = str(e)
                self.file_last_changed[dag.fileloc] = file_last_changed_on_disk
            else:
                found_dags.append(dag)
                found_dags += dag.subdags
        return found_dags
    def bag_dag(self, dag, root_dag):