This is an automated email from the ASF dual-hosted git repository.
potiuk 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 3b9df4f8c93 TimeDeltaSensor - delta templated (#71314)
3b9df4f8c93 is described below
commit 3b9df4f8c93a8fef16b0470863b782d82b613994
Author: raphaelauv <[email protected]>
AuthorDate: Wed Sep 9 14:17:24 2026 +0200
TimeDeltaSensor - delta templated (#71314)
* TimeDeltaSensor - delta templated
* review 1 - add doc
* review 2 - add test on templating error
---------
Co-authored-by: raphaelauv <[email protected]>
---
.../providers/standard/sensors/time_delta.py | 19 ++++++++++----
.../tests/unit/standard/sensors/test_time_delta.py | 30 ++++++++++++++++++++++
2 files changed, 44 insertions(+), 5 deletions(-)
diff --git
a/providers/standard/src/airflow/providers/standard/sensors/time_delta.py
b/providers/standard/src/airflow/providers/standard/sensors/time_delta.py
index 0f9bcf45d07..1295a829bb3 100644
--- a/providers/standard/src/airflow/providers/standard/sensors/time_delta.py
+++ b/providers/standard/src/airflow/providers/standard/sensors/time_delta.py
@@ -48,7 +48,7 @@ class TimeDeltaSensor(BaseSensorOperator):
The delta will be evaluated against data_interval_end if present for the
dag run,
otherwise run_after will be used.
- :param delta: time to wait before succeeding.
+ :param delta: time to wait before succeeding. If a templated value is
provided, it will be cast to an integer representing the duration in minutes.
:param deferrable: Run sensor in deferrable mode. If set to True, task
will defer itself to avoid taking up a worker slot while it is waiting.
.. seealso::
@@ -57,10 +57,12 @@ class TimeDeltaSensor(BaseSensorOperator):
"""
+ template_fields: Sequence[str] = ("delta",)
+
def __init__(
self,
*,
- delta: timedelta,
+ delta: timedelta | int,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
end_from_trigger: bool = False,
**kwargs,
@@ -70,6 +72,12 @@ class TimeDeltaSensor(BaseSensorOperator):
self.deferrable = deferrable
self.end_from_trigger = end_from_trigger
+ def _resolve_delta(self) -> timedelta:
+ value = self.delta
+ if isinstance(value, timedelta):
+ return value
+ return timedelta(minutes=int(value))
+
def _derive_base_time(self, context: Context) -> datetime:
"""
Get the "base time" against which the delta should be calculated.
@@ -93,8 +101,9 @@ class TimeDeltaSensor(BaseSensorOperator):
def poke(self, context: Context) -> bool:
base_time = self._derive_base_time(context=context)
- target_dttm = base_time + self.delta
- self.log.info("Checking if the delta has elapsed base_time=%s,
delta=%s", base_time, self.delta)
+ delta = self._resolve_delta()
+ target_dttm = base_time + delta
+ self.log.info("Checking if the delta has elapsed base_time=%s,
delta=%s", base_time, delta)
return timezone.utcnow() > target_dttm
"""
@@ -113,7 +122,7 @@ class TimeDeltaSensor(BaseSensorOperator):
# Deferrable path
base_time = self._derive_base_time(context=context)
- target_dttm: datetime = base_time + self.delta
+ target_dttm: datetime = base_time + self._resolve_delta()
if timezone.utcnow() > target_dttm:
# If the target datetime is in the past, return immediately
diff --git a/providers/standard/tests/unit/standard/sensors/test_time_delta.py
b/providers/standard/tests/unit/standard/sensors/test_time_delta.py
index 9ec8fc4233d..b0c201583b3 100644
--- a/providers/standard/tests/unit/standard/sensors/test_time_delta.py
+++ b/providers/standard/tests/unit/standard/sensors/test_time_delta.py
@@ -255,6 +255,36 @@ class TestTimeDeltaSensorAsync:
assert caught.value.trigger.moment == expected_time
+ @pytest.mark.parametrize("delta", [timedelta(minutes=1), 1, "{{ 1*2 }}"])
+ def test_time_delta_sensor_templating(self, mocker, delta):
+ defer_mock = mocker.patch(DEFER_PATH)
+ op = TimeDeltaSensor(task_id="time_sensor_check", delta=delta,
dag=self.dag, deferrable=True)
+
+ time = pendulum.datetime(year=2024, month=8, day=1, tz="UTC")
+
+ with time_machine.travel(time, tick=False):
+ data_interval_end = time.add(hours=1)
+
+ context = {"data_interval_end": data_interval_end}
+ op.render_template_fields(context)
+ op.execute(context)
+ defer_mock.assert_called_once()
+
+ def test_time_delta_sensor_templating_error(self, mocker):
+ op = TimeDeltaSensor(
+ task_id="time_sensor_check", delta="{{ 'nothing' }}",
dag=self.dag, deferrable=True
+ )
+ time = pendulum.datetime(year=2024, month=8, day=1, tz="UTC")
+
+ with time_machine.travel(time, tick=False):
+ data_interval_end = time.add(hours=1)
+ context = {"data_interval_end": data_interval_end}
+ op.render_template_fields(context)
+ with pytest.raises(
+ ValueError, match=re.escape("invalid literal for int() with
base 10: 'nothing'")
+ ):
+ op.execute(context)
+
@pytest.mark.parametrize(
"time_to_wait",
[timedelta(minutes=1), 1, "{{ 1*2 }}"],
