Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff 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
17 changes: 11 additions & 6 deletions providers/standard/src/airflow/providers/standard/sensors/time.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,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 @@ -77,14 +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:
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
11 changes: 11 additions & 0 deletions providers/standard/tests/unit/standard/sensors/test_time.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,3 +135,14 @@ 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_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_from_trigger_raises",
schedule=None,
start_date=datetime(2020, 1, 1, 13, 0),
):
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