diff --git a/providers/standard/docs/sensors/datetime.rst b/providers/standard/docs/sensors/datetime.rst index 2a45ecb6bd869..d9949f1ca24c2 100644 --- a/providers/standard/docs/sensors/datetime.rst +++ b/providers/standard/docs/sensors/datetime.rst @@ -41,7 +41,14 @@ TimeSensor Use the :class:`~airflow.providers.standard.sensors.time_sensor.TimeSensor` to end sensing after time specified. ``TimeSensor`` can be run in deferrable mode, if a Triggerer is available. -Time will be evaluated against ``data_interval_end`` if present for the Dag run, otherwise ``run_after`` will be used. +The target moment is computed from the current wall-clock date in the Dag's timezone combined with +``target_time``, evaluated fresh each time the sensor pokes or defers. It is not derived from +``data_interval_end`` or ``run_after``; for interval-relative behavior use +:class:`~airflow.providers.standard.sensors.time_delta.TimeDeltaSensor`. + +``start_from_trigger`` is not supported on ``TimeSensor``: the target moment can only be computed +correctly at task-execution time, not at Dag-parse time, so it cannot be handed to the triggerer in +advance without going stale. Passing ``start_from_trigger=True`` emits a deprecation warning and behaves as if it were not set. .. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_sensors.py :language: python diff --git a/providers/standard/src/airflow/providers/standard/sensors/time.py b/providers/standard/src/airflow/providers/standard/sensors/time.py index 2e3b19eb89940..a69a290d1daa0 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/time.py @@ -17,7 +17,6 @@ # under the License. from __future__ import annotations -import dataclasses import datetime import warnings from typing import TYPE_CHECKING, Any @@ -51,7 +50,6 @@ class TimeSensor(BaseSensorOperator): next_kwargs=None, timeout=None, ) - start_from_trigger = False def __init__( self, @@ -65,31 +63,39 @@ def __init__( ) -> None: super().__init__(**kwargs) - # Create a "date-aware" timestamp that will be used as the "target_datetime". This is a requirement - # of the DateTimeTrigger - - # Get date considering dag.timezone - aware_time = timezone.coerce_datetime( - datetime.datetime.combine( - datetime.datetime.now(self.dag.timezone), target_time, self.dag.timezone - ) - ) - - # Now that the dag's timezone has made the datetime timezone aware, we need to convert to UTC - self.target_datetime = timezone.convert_to_utc(aware_time) + self.target_time = target_time self.deferrable = deferrable self.start_from_trigger = start_from_trigger self.end_from_trigger = end_from_trigger - if self.start_from_trigger: - # Replaced rather than mutated: ``start_trigger_args`` is a class attribute, so - # assigning through it would overwrite the arguments of every other task built - # from this operator. - self.start_trigger_args = dataclasses.replace( - self.start_trigger_args, - trigger_kwargs=dict(moment=self.target_datetime, end_from_trigger=self.end_from_trigger), + @property + def start_from_trigger(self) -> bool: + """Always False: TimeSensor cannot start from trigger. Kept for backward compatibility.""" + return False + + @start_from_trigger.setter + def start_from_trigger(self, value: bool) -> None: + if value: + warnings.warn( + "start_from_trigger is deprecated for TimeSensor and is now ignored. The target " + "moment is computed fresh from the current wall-clock time on every Dag parse, so " + "baking it into the serialized trigger arguments made the serialized Dag hash change " + "on every parse. Use deferrable=True instead, which computes the target moment at " + "task execution time and does not have this problem.", + AirflowProviderDeprecationWarning, + stacklevel=2, ) + @property + def target_datetime(self) -> datetime.datetime: + """Compute the target moment on demand, in the dag's timezone.""" + aware_time = timezone.coerce_datetime( + datetime.datetime.combine( + datetime.datetime.now(self.dag.timezone), self.target_time, self.dag.timezone + ) + ) + return timezone.convert_to_utc(aware_time) + def execute(self, context: Context) -> None: if self.deferrable: self.defer( @@ -106,10 +112,9 @@ def execute_complete(self, context: Context, event: Any = None) -> None: return None def poke(self, context: Context) -> bool: - self.log.info("Checking if the time (%s) has come", self.target_datetime) - - # self.target_date has been converted to UTC, so we do not need to convert timezone - return timezone.utcnow() > self.target_datetime + target_datetime = self.target_datetime + self.log.info("Checking if the time (%s) has come", target_datetime) + return timezone.utcnow() > target_datetime class TimeSensorAsync(TimeSensor): diff --git a/providers/standard/tests/unit/standard/sensors/test_time.py b/providers/standard/tests/unit/standard/sensors/test_time.py index 2b44c61d474b2..cdaf2ed0865e4 100644 --- a/providers/standard/tests/unit/standard/sensors/test_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_time.py @@ -23,6 +23,7 @@ import pytest import time_machine +from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.models.dag import DAG from airflow.providers.common.compat.sdk import TaskDeferred from airflow.providers.standard.sensors.time import TimeSensor @@ -136,21 +137,31 @@ def test_execute_complete_accepts_event(self): except TypeError as e: pytest.fail(f"TypeError raised: {e}") - def test_start_trigger_args_are_not_shared_between_tasks(self): - """Each task must carry its own trigger arguments. - - ``start_trigger_args`` is a class attribute, so assigning through it made every task - built from this operator advertise the moment of whichever was constructed last. - """ + def test_start_from_trigger_is_deprecated_and_ignored(self): + with DAG( + dag_id="test_start_from_trigger_deprecated", + schedule=None, + start_date=datetime(2020, 1, 1, 13, 0), + ): + with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"): + op = TimeSensor(task_id="test", target_time=time(10, 0), start_from_trigger=True) + assert op.start_from_trigger is False + with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"): + op.start_from_trigger = True + assert op.start_from_trigger is False + + def test_target_datetime_recomputed_on_each_access(self): with DAG( - dag_id="test_start_trigger_args_not_shared", + dag_id="test_target_datetime_recomputed", schedule=None, start_date=datetime(2020, 1, 1), ): - early = TimeSensor(task_id="early", target_time=time(1, 0), start_from_trigger=True) - late = TimeSensor(task_id="late", target_time=time(23, 0), start_from_trigger=True) + op = TimeSensor(task_id="test", target_time=time(10, 0)) + + with time_machine.travel("2025-06-01 00:00:00", tick=False): + first = op.target_datetime + with time_machine.travel("2025-06-02 00:00:00", tick=False): + second = op.target_datetime - assert early.start_trigger_args is not late.start_trigger_args - assert early.start_trigger_args.trigger_kwargs["moment"] == early.target_datetime - assert late.start_trigger_args.trigger_kwargs["moment"] == late.target_datetime - assert early.target_datetime != late.target_datetime + assert first.date() == pendulum.datetime(2025, 6, 1).date() + assert second.date() == pendulum.datetime(2025, 6, 2).date()