diff --git a/providers/standard/docs/sensors/datetime.rst b/providers/standard/docs/sensors/datetime.rst index 2a45ecb6bd869..3045a7ffc129f 100644 --- a/providers/standard/docs/sensors/datetime.rst +++ b/providers/standard/docs/sensors/datetime.rst @@ -41,7 +41,7 @@ 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. +Time is evaluated against the wall-clock date in the Dag's timezone at execution time (poke or trigger time), not against ``data_interval_end`` or ``run_after``. For interval-relative behavior, use :class:`~airflow.providers.standard.sensors.time_delta.TimeDeltaSensor`. .. 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 e32a98d97d995..d2ccd9df1b744 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/time.py @@ -24,7 +24,6 @@ from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.common.compat.sdk import BaseSensorOperator, conf, timezone from airflow.providers.standard.triggers.temporal import DateTimeTrigger -from airflow.triggers.base import StartTriggerArgs if TYPE_CHECKING: from airflow.sdk import Context @@ -34,6 +33,11 @@ class TimeSensor(BaseSensorOperator): """ Waits until the specified time of the day. + The time is evaluated against the current date in the Dag's timezone + at execution time (poke or deferral), not at Dag-parse time. This avoids + Dag version churn that previously occurred when target_datetime was + baked into serialized start_trigger_args. + :param target_time: time after which the job succeeds :param deferrable: whether to defer execution @@ -43,53 +47,50 @@ class TimeSensor(BaseSensorOperator): """ - start_trigger_args = StartTriggerArgs( - trigger_cls="airflow.providers.standard.triggers.temporal.DateTimeTrigger", - trigger_kwargs={"moment": "", "end_from_trigger": False}, - next_method="execute_complete", - next_kwargs=None, - timeout=None, - ) - start_from_trigger = False - def __init__( self, *, target_time: datetime.time, deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), - start_from_trigger: bool = False, end_from_trigger: bool = False, trigger_kwargs: dict[str, Any] | None = None, **kwargs, ) -> None: + start_from_trigger = kwargs.pop("start_from_trigger", None) + if start_from_trigger is not None: + warnings.warn( + "start_from_trigger is deprecated and no longer supported for TimeSensor. " + "It has been ignored. Target time is now always evaluated at execution time. " + "Use deferrable=True to defer the sensor instead.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) super().__init__(**kwargs) + self.target_time = target_time + self.deferrable = deferrable + self.end_from_trigger = end_from_trigger + # Accepted for compatibility only; storing it would reintroduce serialized Dag hash churn. + del trigger_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 + def _get_target_datetime(self) -> datetime.datetime: + """Compute target datetime at execution time, not parse time.""" + dag_timezone = getattr(getattr(self, "dag", None), "timezone", None) or timezone.utc + now_date = datetime.datetime.now(dag_timezone).date() aware_time = timezone.coerce_datetime( - datetime.datetime.combine( - datetime.datetime.now(self.dag.timezone), target_time, self.dag.timezone - ) + datetime.datetime.combine(now_date, self.target_time, dag_timezone) ) + return timezone.convert_to_utc(aware_time) - # 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.deferrable = deferrable - self.start_from_trigger = start_from_trigger - self.end_from_trigger = end_from_trigger - - if self.start_from_trigger: - self.start_trigger_args.trigger_kwargs = dict( - moment=self.target_datetime, end_from_trigger=self.end_from_trigger - ) + @property + def target_datetime(self) -> datetime.datetime: + """Backward-compatible property, now computed at call time.""" + return self._get_target_datetime() def execute(self, context: Context) -> None: if self.deferrable: self.defer( trigger=DateTimeTrigger( - moment=self.target_datetime, # This needs to be an aware timestamp + moment=self._get_target_datetime(), end_from_trigger=self.end_from_trigger, ), method_name="execute_complete", @@ -101,10 +102,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._get_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 1989c35d91b58..35707e8524d27 100644 --- a/providers/standard/tests/unit/standard/sensors/test_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_time.py @@ -23,10 +23,12 @@ import pytest import time_machine -from airflow.models.dag import DAG +from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.common.compat.sdk import TaskDeferred from airflow.providers.standard.sensors.time import TimeSensor from airflow.providers.standard.triggers.temporal import DateTimeTrigger +from airflow.sdk import DAG +from airflow.serialization.serialized_objects import LazyDeserializedDAG from tests_common.test_utils.compat import timezone @@ -135,3 +137,37 @@ def test_execute_complete_accepts_event(self): op.execute_complete(context={}, event={"status": "success"}) except TypeError as e: pytest.fail(f"TypeError raised: {e}") + + def test_start_from_trigger_is_ignored_with_warning(self): + with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"): + op = TimeSensor( + task_id="test", + target_time=time(10, 0), + start_from_trigger=True, + trigger_kwargs={"moment": "2020-01-01T10:00:00Z"}, + ) + + assert getattr(op, "start_from_trigger", False) is False + assert not hasattr(op, "_trigger_kwargs") + + def test_ignored_start_from_trigger_keeps_serialized_dag_hash_stable(self): + def build_hash() -> str: + with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"): + with DAG( + dag_id="test_ignored_start_from_trigger_hash", + schedule=None, + start_date=datetime(2020, 1, 1), + ) as dag: + TimeSensor( + task_id="test", + target_time=time(10, 0), + start_from_trigger=True, + ) + return LazyDeserializedDAG.from_dag(dag).hash + + with time_machine.travel("2025-01-01 00:00:00", tick=False): + first_hash = build_hash() + with time_machine.travel("2025-01-02 00:00:00", tick=False): + second_hash = build_hash() + + assert first_hash == second_hash