This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""
        Test that if a DAG does not exist in serialized_dag table (as the DAG file was removed),
        remove dags from the DagBag
        """
        from airflow.operators.empty import EmptyOperator
        with dag_maker(
            dag_id="test_dag_removed_if_serialized_dag_is_removed",
            schedule=None,
            start_date=tz.datetime(2021, 10, 12),
        ) as dag:
            EmptyOperator(task_id="task_1")
        dag_maker.create_dagrun()
        dagbag = DagBag(dag_folder=self.empty_dir, include_examples=False, read_dags_from_db=True)
        dagbag.dags = {dag.dag_id: SerializedDAG.from_dict(SerializedDAG.to_dict(dag))}
        dagbag.dags_last_fetched = {dag.dag_id: (tz.utcnow() - timedelta(minutes=2))}
        dagbag.dags_hash = {dag.dag_id: mock.ANY}
        assert SerializedDagModel.has_dag(dag.dag_id) is False
        assert dagbag.get_dag(dag.dag_id) is None
        assert dag.dag_id not in dagbag.dags
        assert dag.dag_id not in dagbag.dags_last_fetched
        assert dag.dag_id not in dagbag.dags_hash
    def process_dag(self, create_dag):