This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
if recursive:
                for subdag in subdags:
                    subdag.fileloc = dag.fileloc
                    subdag.parent_dag = dag
                    self._bag_dag(dag=subdag, root_dag=root_dag, recursive=False)
            prev_dag = self.dags.get(dag.dag_id)
            if prev_dag and prev_dag.fileloc != dag.fileloc:
                raise AirflowDagDuplicatedIdException(
                    dag_id=dag.dag_id,
                    incoming=dag.fileloc,
                    existing=self.dags[dag.dag_id].fileloc,
                )
            self.dags[dag.dag_id] = dag
            self.log.debug("Loaded DAG %s", dag)
        except (AirflowDagCycleException, AirflowDagDuplicatedIdException):
            
            self.log.exception("Exception bagging dag: %s", dag.dag_id)
            
            
            if recursive:
                for subdag in subdags:
                    if subdag.dag_id in self.dags:
                        del self.dags[subdag.dag_id]
            raise
    def collect_dags(
        self,
        dag_folder: str | pathlib.Path | None = None,
        only_if_updated: bool = True,