diff --git a/providers/standard/src/airflow/providers/standard/sensors/external_task.py b/providers/standard/src/airflow/providers/standard/sensors/external_task.py index 8270386754f0a..034e86ca2612c 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/external_task.py +++ b/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, diff --git a/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py b/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py index ef6bd33a76423..3bf470ae33b49 100644 --- a/providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py +++ b/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(