Migrate dbt.cloud provider to common.compat (apache#56999)

Replace version-specific conditional imports with ``common.compat`` layer.
This standardizes compatibility handling across Airflow 2.x and 3.x.

Author: kaxil

providers/dbt/cloud/pyproject.toml

@@ -58,7 +58,7 @@ requires-python = ">=3.10"
 # After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build``
 dependencies = [
     "apache-airflow>=2.10.0",
-    "apache-airflow-providers-common-compat>=1.6.0",
+    "apache-airflow-providers-common-compat>=1.7.4",    # + TODO: bump to next version
     "apache-airflow-providers-http",
     "asgiref>=2.3.0",
     "aiohttp>=3.9.2",

providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py

@@ -24,6 +24,7 @@
 from typing import TYPE_CHECKING, Any
 
 from airflow.configuration import conf
+from airflow.providers.common.compat.sdk import BaseOperator, BaseOperatorLink, XCom
 from airflow.providers.dbt.cloud.hooks.dbt import (
     DbtCloudHook,
     DbtCloudJobRunException,
@@ -32,11 +33,6 @@
 )
 from airflow.providers.dbt.cloud.triggers.dbt import DbtCloudRunJobTrigger
 from airflow.providers.dbt.cloud.utils.openlineage import generate_openlineage_events_from_dbt_cloud_run
-from airflow.providers.dbt.cloud.version_compat import (
-    BaseOperator,
-    BaseOperatorLink,
-    XCom,
-)
 
 if TYPE_CHECKING:
     from airflow.providers.openlineage.extractors import OperatorLineage

providers/dbt/cloud/src/airflow/providers/dbt/cloud/sensors/dbt.py

@@ -22,15 +22,10 @@
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
+from airflow.providers.common.compat.sdk import BaseSensorOperator
 from airflow.providers.dbt.cloud.hooks.dbt import DbtCloudHook, DbtCloudJobRunException, DbtCloudJobRunStatus
 from airflow.providers.dbt.cloud.triggers.dbt import DbtCloudRunJobTrigger
 from airflow.providers.dbt.cloud.utils.openlineage import generate_openlineage_events_from_dbt_cloud_run
-from airflow.providers.dbt.cloud.version_compat import AIRFLOW_V_3_0_PLUS
-
-if AIRFLOW_V_3_0_PLUS:
-    from airflow.sdk import BaseSensorOperator
-else:
-    from airflow.sensors.base import BaseSensorOperator  # type: ignore[no-redef]
 
 if TYPE_CHECKING:
     from airflow.providers.openlineage.extractors import OperatorLineage

providers/dbt/cloud/src/airflow/providers/dbt/cloud/version_compat.py

@@ -34,18 +34,6 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 
 AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
 
-if AIRFLOW_V_3_0_PLUS:
-    from airflow.sdk import BaseOperator, BaseOperatorLink, BaseSensorOperator
-    from airflow.sdk.execution_time.xcom import XCom
-else:
-    from airflow.models import BaseOperator, XCom
-    from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
-    from airflow.sensors.base import BaseSensorOperator  # type: ignore[no-redef]
-
 __all__ = [
     "AIRFLOW_V_3_0_PLUS",
-    "BaseOperator",
-    "BaseSensorOperator",
-    "BaseOperatorLink",
-    "XCom",
 ]

providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py

@@ -24,14 +24,14 @@
 
 from airflow.exceptions import TaskDeferred
 from airflow.models import DAG, Connection
+from airflow.providers.common.compat.sdk import timezone
 from airflow.providers.dbt.cloud.hooks.dbt import DbtCloudHook, DbtCloudJobRunException, DbtCloudJobRunStatus
 from airflow.providers.dbt.cloud.operators.dbt import (
     DbtCloudGetJobRunArtifactOperator,
     DbtCloudListJobsOperator,
     DbtCloudRunJobOperator,
 )
 from airflow.providers.dbt.cloud.triggers.dbt import DbtCloudRunJobTrigger
-from airflow.utils import timezone
 
 from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS