Run providers compat test with Airflow 3.1 & fix compat (apache#56118)

Some compat layers were wrong! Fixed them while adding 3.1.0 to provider compat matrix

Author: kaxil

dev/breeze/src/airflow_breeze/global_constants.py

@@ -793,6 +793,12 @@ def generate_provider_dependencies_if_needed():
         "remove-providers": "",
         "run-unit-tests": "true",
     },
+    {
+        "python-version": "3.10",
+        "airflow-version": "3.1.0",
+        "remove-providers": "",
+        "run-unit-tests": "true",
+    },
 ]
 
 ALL_PYTHON_VERSION_TO_PATCHLEVEL_VERSION: dict[str, str] = {

devel-common/src/tests_common/pytest_plugin.py

@@ -1175,7 +1175,10 @@ def run_ti(self, task_id, dag_run=None, dag_run_kwargs=None, map_index=-1, **kwa
 
         def sync_dagbag_to_db(self):
             if AIRFLOW_V_3_1_PLUS:
-                from airflow.dag_processing.dagbag import sync_bag_to_db
+                try:
+                    from airflow.dag_processing.dagbag import sync_bag_to_db
+                except ImportError:
+                    from airflow.models.dagbag import sync_bag_to_db
 
                 sync_bag_to_db(self.dagbag, self.bundle_name, None)
             elif AIRFLOW_V_3_0_PLUS:

devel-common/src/tests_common/test_utils/db.py

@@ -105,8 +105,11 @@ def _bootstrap_dagbag():
 
         dagbag = DagBag()
         # Save DAGs in the ORM
-        if AIRFLOW_V_3_2_PLUS:
-            from airflow.dag_processing.dagbag import sync_bag_to_db
+        if AIRFLOW_V_3_1_PLUS:
+            try:
+                from airflow.dag_processing.dagbag import sync_bag_to_db
+            except ImportError:
+                from airflow.models.dagbag import sync_bag_to_db
 
             sync_bag_to_db(dagbag, bundle_name="dags-folder", bundle_version=None, session=session)
         elif AIRFLOW_V_3_0_PLUS:
@@ -180,8 +183,11 @@ def parse_and_sync_to_db(folder: Path | str, include_examples: bool = False):
             session.flush()
 
         dagbag = DagBag(dag_folder=folder, include_examples=include_examples)
-        if AIRFLOW_V_3_2_PLUS:
-            from airflow.dag_processing.dagbag import sync_bag_to_db
+        if AIRFLOW_V_3_1_PLUS:
+            try:
+                from airflow.dag_processing.dagbag import sync_bag_to_db
+            except ImportError:
+                from airflow.models.dagbag import sync_bag_to_db  # type: ignore[no-redef, attribute-defined]
 
             sync_bag_to_db(dagbag, "dags-folder", None, session=session)
         elif AIRFLOW_V_3_0_PLUS: