Deprecating poll_interval in ExternalTaskSensor for poke_interval

providers/standard/src/airflow/providers/standard/sensors/external_task.py

@@ -23,6 +23,7 @@
 from collections.abc import Callable, Collection, Iterable, Sequence
 from typing import TYPE_CHECKING, ClassVar
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.models.dag import DagModel
 from airflow.providers.common.compat.sdk import (
     AirflowSkipException,
@@ -170,7 +171,8 @@ class ExternalTaskSensor(BaseSensorOperator):
         external_task_id is not None) or check if the DAG to wait for exists (when
         external_task_id is None), and immediately cease waiting if the external task
         or DAG does not exist (default value: False).
-    :param poll_interval: polling period in seconds to check for the status
+    :param poke_interval: polling period in seconds to check for the status
+    :param poll_interval: (DEPRECATED) use ``poke_interval`` instead
     :param deferrable: Run sensor in deferrable mode
     """
 
@@ -191,10 +193,19 @@ def __init__(
         execution_delta: datetime.timedelta | None = None,
         execution_date_fn: Callable | None = None,
         check_existence: bool = False,
-        poll_interval: float = 2.0,
+        poll_interval: float | None = None,
         deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
         **kwargs,
     ):
+        if poll_interval is not None:
+            warnings.warn(
+                "Parameter `poll_interval` is deprecated and will be removed in a future release. "
+                "Use `poke_interval` instead.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
+            )
+            kwargs.setdefault("poke_interval", poll_interval)
+
         super().__init__(**kwargs)
 
         self.allowed_states: list[str] = (
@@ -263,9 +274,29 @@ def __init__(
         self.check_existence = check_existence
         self._has_checked_existence = False
         self.deferrable = deferrable
-        self.poll_interval = poll_interval
         self.external_dates_filter: str | None = None
 
+    @property
+    def poll_interval(self) -> float:
+        """Deprecated alias for poke_interval."""
+        warnings.warn(
+            "Attribute poll_interval is deprecated and will be removed in a future release. "
+            "Use poke_interval instead.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
+        return self.poke_interval
+
+    @poll_interval.setter
+    def poll_interval(self, value: float) -> None:
+        warnings.warn(
+            "Attribute poll_interval is deprecated and will be removed in a future release. "
+            "Use poke_interval instead.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
+        self.poke_interval = value
+
     def _get_dttm_filter(self, context: Context) -> Sequence[datetime.datetime]:
         logical_date = self._get_logical_date(context)
 
@@ -452,7 +483,7 @@ def execute(self, context: Context) -> None:
                         allowed_states=self.allowed_states,
                         failed_states=self.failed_states,
                         skipped_states=self.skipped_states,
-                        poke_interval=self.poll_interval,
+                        poke_interval=self.poke_interval,
                         soft_fail=self.soft_fail,
                         logical_dates=list(dttm_filter),
                         run_ids=None,
@@ -477,7 +508,7 @@ def execute(self, context: Context) -> None:
                         allowed_states=self.allowed_states,
                         failed_states=self.failed_states,
                         skipped_states=self.skipped_states,
-                        poke_interval=self.poll_interval,
+                        poke_interval=self.poke_interval,
                         soft_fail=self.soft_fail,
                         execution_dates=list(dttm_filter),
                         logical_dates=None,

providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py

@@ -27,6 +27,7 @@
 from sqlalchemy import select
 
 from airflow import settings
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.models import DagRun, TaskInstance
 from airflow.models.dag import DAG
 from airflow.models.serialized_dag import SerializedDagModel
@@ -1445,6 +1446,121 @@ def test_external_task_sensor_deferrable(self, dag_maker):
         assert exc.value.trigger.external_task_ids == ["test_task"]
         assert exc.value.trigger.logical_dates == [DEFAULT_DATE]
 
+    def test_poke_interval_set_on_init(self):
+        """Test that poke_interval is set on init and the deprecated poll_interval attribute mirrors it."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+            poke_interval=30,
+        )
+
+        assert sensor.poke_interval == 30
+        with pytest.warns(AirflowProviderDeprecationWarning, match="poll_interval"):
+            assert sensor.poll_interval == 30
+
+    def test_poll_interval_attribute_get_set_deprecated(self):
+        """Reading or writing the poll_interval attribute is deprecated but still mirrors poke_interval."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+            poke_interval=30,
+        )
+
+        with pytest.warns(AirflowProviderDeprecationWarning, match="poll_interval"):
+            sensor.poll_interval = 15
+
+        assert sensor.poke_interval == 15
+
+    def test_poke_interval_default_when_unset(self):
+        """The BaseSensor default of 60.0 must be preserved."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+        )
+
+        assert sensor.poke_interval == 60.0
+
+    def test_poke_interval_accepts_timedelta(self):
+        """poke_interval should accept a timedelta, coerced to seconds, same as the base sensor."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+            poke_interval=timedelta(seconds=5),
+        )
+
+        assert sensor.poke_interval == 5.0
+
+    def test_poke_interval_no_warning_when_poll_interval_unset(self, recwarn):
+        """Passing only poke_interval must not emit the poll_interval deprecation warning."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+            poke_interval=15,
+        )
+
+        assert sensor.poke_interval == 15
+        assert not any(issubclass(w.category, AirflowProviderDeprecationWarning) for w in recwarn.list)
+
+    def test_poll_interval_deprecated_and_sets_poke_interval(self):
+        """poll_interval is deprecated, but its value is still respected as poke_interval."""
+        with pytest.warns(AirflowProviderDeprecationWarning, match="poll_interval"):
+            sensor = ExternalTaskSensor(
+                task_id=TASK_ID,
+                external_task_id=EXTERNAL_TASK_ID,
+                external_dag_id=EXTERNAL_DAG_ID,
+                poll_interval=45,
+            )
+
+        assert sensor.poke_interval == 45
+
+    def test_poll_interval_zero_still_deprecated(self):
+        """Regression test: a falsy poll_interval (0) must still take the deprecation path."""
+        with pytest.warns(AirflowProviderDeprecationWarning, match="poll_interval"):
+            sensor = ExternalTaskSensor(
+                task_id=TASK_ID,
+                external_task_id=EXTERNAL_TASK_ID,
+                external_dag_id=EXTERNAL_DAG_ID,
+                poll_interval=0,
+            )
+
+        assert sensor.poke_interval == 0
+
+    def test_poke_interval_overrides_poll_interval_when_both_set(self):
+        """When both are provided, poke_interval takes precedence over the deprecated poll_interval."""
+        with pytest.warns(AirflowProviderDeprecationWarning, match="poll_interval"):
+            sensor = ExternalTaskSensor(
+                task_id=TASK_ID,
+                external_task_id=EXTERNAL_TASK_ID,
+                external_dag_id=EXTERNAL_DAG_ID,
+                poke_interval=10,
+                poll_interval=99,
+            )
+
+        assert sensor.poke_interval == 10
+
+    @pytest.mark.execution_timeout(10)
+    def test_deferrable_poke_interval_passed_to_trigger(self, dag_maker):
+        """Test that poke_interval is correctly forwarded to WorkflowTrigger when deferrable=True."""
+        with dag_maker("test_dag_child"):
+            op = ExternalTaskSensor(
+                task_id="test_external_task_sensor_check",
+                external_dag_id="test_dag_parent",
+                external_task_id="test_task",
+                deferrable=True,
+                poke_interval=30,
+            )
+
+        with pytest.raises(TaskDeferred) as exc:
+            op.execute(context=self.context)
+
+        assert isinstance(exc.value.trigger, WorkflowTrigger)
+        assert exc.value.trigger.poke_interval == 30
+
     @pytest.mark.execution_timeout(10)
     def test_external_task_sensor_only_dag_id(self, dag_maker):
         """Test that the sensor works correctly when only external_dag_id is provided."""
@@ -1550,6 +1666,22 @@ def test_defer_and_fire_task_state_trigger(self):
 
         assert isinstance(exc.value.trigger, WorkflowTrigger), "Trigger is not a WorkflowTrigger"
 
+    def test_deferrable_poke_interval_passed_to_trigger(self):
+        """Test that poke_interval flows through to WorkflowTrigger on both AF2 and AF3 paths."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+            deferrable=True,
+            poke_interval=30,
+        )
+
+        with pytest.raises(TaskDeferred) as exc:
+            sensor.execute(context={"execution_date": DEFAULT_DATE, "logical_date": DEFAULT_DATE})
+
+        assert isinstance(exc.value.trigger, WorkflowTrigger)
+        assert exc.value.trigger.poke_interval == 30
+
     def test_defer_and_fire_failed_state_trigger(self):
         """Tests that an ExternalTaskNotFoundError is raised in case of error event"""
         sensor = ExternalTaskSensor(
@@ -1664,6 +1796,19 @@ def test_defer_execute_complete_re_sets_external_dates_filter_attr(self):
 
         assert sensor.external_dates_filter == DEFAULT_DATE.isoformat()
 
+    def test_poke_interval_set_on_init(self):
+        """Test that poke_interval is set on init and the deprecated poll_interval attribute mirrors it."""
+        sensor = ExternalTaskSensor(
+            task_id=TASK_ID,
+            external_task_id=EXTERNAL_TASK_ID,
+            external_dag_id=EXTERNAL_DAG_ID,
+            poke_interval=30,
+        )
+
+        assert sensor.poke_interval == 30
+        with pytest.warns(AirflowProviderDeprecationWarning, match="poll_interval"):
+            assert sensor.poll_interval == 30
+
 
 @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Needs Flask app context fixture for AF 2")
 @pytest.mark.parametrize(