This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
self.dags = {
                key: dag
                for key, dag in self.dags.items()
                if root_dag_id != key and not (dag.parent_dag and root_dag_id == dag.parent_dag.dag_id)
            }
        if is_missing or is_expired:
            
            found_dags = self.process_file(
                filepath=correct_maybe_zipped(orm_dag.fileloc), only_if_updated=False
            )
            
            if found_dags and dag_id in [found_dag.dag_id for found_dag in found_dags]:
                return self.dags[dag_id]
            elif dag_id in self.dags:
                del self.dags[dag_id]
        return self.dags.get(dag_id)
    def _add_dag_from_db(self, dag_id: str, session: Session):
        """Add DAG to DagBag from DB"""
        from airflow.models.serialized_dag import SerializedDagModel
        row = SerializedDagModel.get(dag_id, session)
        if not row:
            return None
        row.load_op_links = self.load_op_links
        dag = row.dag
        for subdag in dag.subdags:
            self.dags[subdag.dag_id] = subdag
        self.dags[dag.dag_id] = dag
        self.dags_last_fetched[dag.dag_id] = timezone.utcnow()
        self.dags_hash[dag.dag_id] = row.dag_hash
    def process_file(self, filepath, only_if_updated=True, safe_mode=True):