Skip to content

Fix TimeSensor start_from_trigger causing dag_version churn#69610

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bramhanandlingala wants to merge 6 commits into
base:main
Choose a base branch
from
Open

Fix TimeSensor start_from_trigger causing dag_version churn #69610

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9bf1379
Fix TimeSensor start_from_trigger causing dag_version churn
bramhanandlingala Jul 8, 2026
4edcc55
Merge branch 'main' into fix/#69543
bramhanandlingala Jul 16, 2026
ff49e68
Merge branch 'main' into fix/#69543
bramhanandlingala Jul 19, 2026
cae1d1a
Merge branch 'main' into fix/#69543
bramhanandlingala Jul 24, 2026
d03ea3d
Merge branch 'main' into fix/#69543
bramhanandlingala Aug 12, 2026
be5af30
Merge branch 'main' into fix/#69543
bramhanandlingala Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion providers/standard/docs/sensors/datetime.rst
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -41,7 +41,13 @@ 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 "today" (in the Dag's timezone) combined with ``target_time``, evaluated fresh
each time the operator is instantiated -- which, for the default and for ``deferrable=True`` execution,
happens close to the actual task run.

``start_from_trigger`` is not supported by ``TimeSensor``: the target moment can only be correctly
computed at task-execution time, not at Dag-parse time, so it cannot be handed to the triggerer ahead
of time. Setting ``start_from_trigger=True`` raises a ``ValueError``.

.. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_sensors.py
:language: python
Expand Down
22 changes: 11 additions & 11 deletions providers/standard/src/airflow/providers/standard/sensors/time.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,7 +17,6 @@
# under the License.
from __future__ import annotations

import dataclasses
import datetime
import warnings
from typing import TYPE_CHECKING, Any
Expand DownExpand Up@@ -65,6 +64,16 @@ def __init__(
) -> None:
super().__init__(**kwargs)

if start_from_trigger:
raise ValueError(
"TimeSensor does not support start_from_trigger=True: the target "
"datetime is computed from the current wall-clock time at DAG-parse "
"time, not from the DagRun's data_interval_end/run_after, so baking "
"it into the serialized trigger arguments causes the DAG version to "
"change on every parse. Use deferrable=True without start_from_trigger, "
"or use DateTimeSensor with a templated target_time instead."
)

# Create a "date-aware" timestamp that will be used as the "target_datetime". This is a requirement
# of the DateTimeTrigger

Expand All@@ -78,18 +87,9 @@ def __init__(
# 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.start_from_trigger = False
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),
)

def execute(self, context: Context) -> None:
if self.deferrable:
self.defer(
Expand Down
22 changes: 7 additions & 15 deletions providers/standard/tests/unit/standard/sensors/test_time.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -136,21 +136,13 @@ 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_raises_value_error(self):
"""start_from_trigger bakes datetime.now() into trigger_kwargs at parse time,
causing the DAG version to change on every parse. See #69543."""
with DAG(
dag_id="test_start_trigger_args_not_shared",
dag_id="test_start_from_trigger_raises",
schedule=None,
start_date=datetime(2020, 1, 1),
start_date=datetime(2020, 1, 1, 13, 0),
):
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)

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
with pytest.raises(ValueError, match="TimeSensor does not support start_from_trigger=True"):
TimeSensor(task_id="test", target_time=time(10, 0), start_from_trigger=True)
Loading