This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
file_parse_end_dttm = timezone.utcnow()
                stats.append(
                    FileLoadStat(
                        file=filepath.replace(settings.DAGS_FOLDER, ""),
                        duration=file_parse_end_dttm - file_parse_start_dttm,
                        dag_num=len(found_dags),
                        task_num=sum(len(dag.tasks) for dag in found_dags),
                        dags=str([dag.dag_id for dag in found_dags]),
                    )
                )
            except Exception as e:
                self.log.exception(e)
        self.dagbag_stats = sorted(stats, key=lambda x: x.duration, reverse=True)
    def collect_dags_from_db(self):
        """Collects DAGs from database."""
        from airflow.models.serialized_dag import SerializedDagModel
        with Stats.timer("collect_db_dags"):
            self.log.info("Filling up the DagBag from database")
            
            
            self.dags = SerializedDagModel.read_all_dags()
            
            
            
            subdags = {}
            for dag in self.dags.values():
                for subdag in dag.subdags:
                    subdags[subdag.dag_id] = subdag
            self.dags.update(subdags)
    def dagbag_report(self):