Centralize version-aware EmptyOperator imports in a compatibility module

cosmos/airflow/compatibility.py

@@ -0,0 +1,64 @@
+"""Version-aware imports for Airflow objects whose import path differs across Airflow 2 and 3.
+
+Names exported here are resolved lazily on first attribute access (PEP 562), so importing
+this module is free: the underlying Airflow object is only imported when the name is actually
+used. Use sites keep referencing the object by its real name (e.g. ``EmptyOperator``) rather
+than a version-specific dotted path.
+"""
+
+from __future__ import annotations
+
+import importlib
+from typing import TYPE_CHECKING
+
+from cosmos.constants import _AIRFLOW3_MAJOR_VERSION, AIRFLOW_VERSION
+
+if TYPE_CHECKING:
+    # Resolved for type checkers / IDEs only; no import happens at runtime here. The standard
+    # provider path is tried first so the type resolves to the real class on Airflow 3 (the
+    # legacy module is a deprecated shim typed as ``Any`` there).
+    try:
+        from airflow.providers.standard.operators.empty import EmptyOperator as EmptyOperator
+    except ImportError:
+        from airflow.operators.empty import EmptyOperator as EmptyOperator  # type: ignore[no-redef]
+
+# Single source of truth for where ``EmptyOperator`` lives. The operator moved to the standard
+# provider in Airflow 3; the legacy ``airflow.operators.empty`` path still resolves there but
+# emits a ``DeprecatedImportWarning``, so on Airflow 3 we select the standard provider path.
+# Compare on the major version so Airflow 3 pre-releases (e.g. 3.0.0rc1) are treated as Airflow 3.
+_EMPTY_OPERATOR_MODULE = (
+    "airflow.operators.empty"
+    if AIRFLOW_VERSION.major < _AIRFLOW3_MAJOR_VERSION
+    else "airflow.providers.standard.operators.empty"
+)
+
+# Maps the exported name to the module it should be imported from for this Airflow version.
+_LAZY_IMPORTS = {
+    "EmptyOperator": _EMPTY_OPERATOR_MODULE,
+}
+
+
+def __getattr__(name: str) -> object:
+    module_path = _LAZY_IMPORTS.get(name)
+    if module_path is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    # Cache the resolved symbol in the module namespace so subsequent attribute access skips
+    # __getattr__ entirely (PEP 562 only invokes it for names missing from globals()).
+    resolved = getattr(importlib.import_module(module_path), name)
+    globals()[name] = resolved
+    return resolved
+
+
+def get_version_aware_operator_class_path(operator: type) -> str:
+    """Return the fully qualified import path for the given operator class.
+
+    The path is read from the class itself (``__module__`` + ``__name__``), so it reflects
+    the actual module the class lives in for the running Airflow version. Pair it with the
+    version-aware classes exported from this module, e.g.::
+
+        get_version_aware_operator_class_path(EmptyOperator)
+
+    Used where Cosmos stores the operator as a dotted string for later dynamic import
+    (e.g. ``Task.operator_class``) instead of referencing the class directly.
+    """
+    return f"{operator.__module__}.{operator.__name__}"

cosmos/airflow/graph.py

@@ -24,6 +24,7 @@
     from airflow.utils.task_group import TaskGroup
 
 from cosmos import settings
+from cosmos.airflow.compatibility import EmptyOperator, get_version_aware_operator_class_path
 from cosmos.config import ExecutionConfig, RenderConfig
 from cosmos.constants import (
     DBT_SETUP_ASYNC_TASK_ID,
@@ -417,7 +418,11 @@ def create_task_metadata(  # noqa: C901
                     args = {"task_display_name": args["task_display_name"]}
                 else:
                     args = {}
-                return TaskMetadata(id=task_id, operator_class="airflow.operators.empty.EmptyOperator", arguments=args)
+                return TaskMetadata(
+                    id=task_id,
+                    operator_class=get_version_aware_operator_class_path(EmptyOperator),
+                    arguments=args,
+                )
         else:  # DbtResourceType.MODEL, DbtResourceType.SEED and DbtResourceType.SNAPSHOT
             if node.fqn and len(node.fqn) > 0:
                 args[models_select_key] = f"fqn:{'.'.join(node.fqn)}"

cosmos/operators/_watcher/base.py

@@ -51,7 +51,7 @@
         from airflow.sdk import DAG
     except ImportError:
         from airflow.models.dag import DAG  # type: ignore[assignment]
-    from airflow.operators.empty import EmptyOperator
+    from cosmos.airflow.compatibility import EmptyOperator
 
     try:
         from airflow.sdk import TaskGroup
@@ -779,10 +779,7 @@ def create_producer_done_task(dag: DAG, task_group: TaskGroup, task_id: str) ->
     is skipped on retry, this task still succeeds (trigger_rule=NONE_FAILED), preventing
     the skip from propagating to tasks downstream of the group.
     """
-    try:
-        from airflow.providers.standard.operators.empty import EmptyOperator
-    except ImportError:
-        from airflow.operators.empty import EmptyOperator  # type: ignore[no-redef]
+    from cosmos.airflow.compatibility import EmptyOperator
 
     try:
         from airflow.task.trigger_rule import TriggerRule

cosmos/operators/watcher_kubernetes.py

@@ -16,12 +16,8 @@
 from airflow.exceptions import AirflowException, AirflowSkipException
 from airflow.providers.cncf.kubernetes.callbacks import KubernetesPodOperatorCallback, client_type
 
-try:
-    from airflow.providers.standard.operators.empty import EmptyOperator
-except ImportError:  # pragma: no cover
-    from airflow.operators.empty import EmptyOperator  # type: ignore[no-redef]
-
 from cosmos.airflow._override import CosmosKubernetesPodManager
+from cosmos.airflow.compatibility import EmptyOperator
 from cosmos.log import get_logger
 from cosmos.operators._watcher.base import BaseConsumerSensor, store_dbt_resource_status_from_log
 from cosmos.operators._watcher.xcom import (

tests/airflow/test_graph.py

@@ -18,6 +18,7 @@
     from airflow.operators.empty import EmptyOperator
     from airflow.utils.task_group import TaskGroup
 
+from cosmos.airflow.compatibility import get_version_aware_operator_class_path
 from cosmos.airflow.graph import (
     _add_teardown_task,
     _add_watcher_producer_task,
@@ -691,7 +692,7 @@ def test_create_task_metadata_model_use_task_group(caplog):
             False,
             SOURCE_RENDERING_BEHAVIOR,
             "my_source_source",
-            "airflow.operators.empty.EmptyOperator",
+            get_version_aware_operator_class_path(EmptyOperator),
         ),
         (
             f"{DbtResourceType.SOURCE.value}.my_folder.my_source",