This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 36e7d02a40c Deprecating `poll_interval` in `ExternalTaskSensor` for
`poke_interval` (#68997)
36e7d02a40c is described below
commit 36e7d02a40c5d1117ebbbd36f2af5737b25ada24
Author: Jake McGrath <[email protected]>
AuthorDate: Fri Jul 31 09:27:59 2026 -0400
Deprecating `poll_interval` in `ExternalTaskSensor` for `poke_interval`
(#68997)
Co-authored-by: Phani Kumar <[email protected]>
Co-authored-by: Kaxil Naik <[email protected]>
Co-authored-by: Tzu-ping Chung <[email protected]>
---
.../providers/standard/sensors/external_task.py | 41 +++++-
.../standard/sensors/test_external_task_sensor.py | 145 +++++++++++++++++++++
2 files changed, 181 insertions(+), 5 deletions(-)
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 8270386754f..034e86ca261 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 @@ import warnings
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 @@ class ExternalTaskSensor(BaseSensorOperator):
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 @@ class ExternalTaskSensor(BaseSensorOperator):
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 @@ class ExternalTaskSensor(BaseSensorOperator):
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 @@ class ExternalTaskSensor(BaseSensorOperator):
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 ef6bd33a764..3bf470ae33b 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 @@ import pytest
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 @@ class TestExternalTaskSensorV3:
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 @@ class TestExternalTaskAsyncSensor:
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 @@ class TestExternalTaskAsyncSensor:
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(
