This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""DAGs are collected from Database"""
        db.clear_db_dags()
        dagbag = DagBag(str(example_dags_folder))
        example_dags = dagbag.dags
        for dag in example_dags.values():
            SerializedDagModel.write_dag(dag)
        new_dagbag = DagBag(read_dags_from_db=True)
        assert len(new_dagbag.dags) == 0
        new_dagbag.collect_dags_from_db()
        new_dags = new_dagbag.dags
        assert len(example_dags) == len(new_dags)
        for dag_id, dag in example_dags.items():
            serialized_dag = new_dags[dag_id]
            assert serialized_dag.dag_id == dag.dag_id
            assert set(serialized_dag.task_dict) == set(dag.task_dict)
    @patch("airflow.settings.task_policy", cluster_policies.example_task_policy)
    def test_task_cluster_policy_violation(self):