From ba1c37ce66eadca41f919f01354230342437603a Mon Sep 17 00:00:00 2001 From: haseebmalik18 Date: Thu, 23 Jul 2026 13:27:59 -0400 Subject: [PATCH] Fix DateTimeSensorAsync crashing Dag parsing with templated target_time --- .../providers/standard/sensors/date_time.py | 11 +++++- .../unit/standard/sensors/test_date_time.py | 37 ++++++++++++++++++- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/providers/standard/src/airflow/providers/standard/sensors/date_time.py b/providers/standard/src/airflow/providers/standard/sensors/date_time.py index e0b64fbf51838..4e26e6c63afdf 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/date_time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/date_time.py @@ -98,6 +98,9 @@ class DateTimeSensorAsync(DateTimeSensor): :param target_time: datetime after which the job succeeds. (templated) :param start_from_trigger: Start the task directly from the triggerer without going into the worker. + This requires a static ``target_time`` (a datetime or ISO-8601 string). A templated + ``target_time`` is not supported here because the trigger is created at Dag-parse time, + before Jinja templates are rendered. :param trigger_kwargs: The keyword arguments passed to the trigger when start_from_trigger is set to True during dynamic task mapping. This argument is not used in standard usage. :param end_from_trigger: End the task directly from the triggerer without going into the worker. @@ -125,8 +128,14 @@ def __init__( self.start_from_trigger = start_from_trigger if self.start_from_trigger: + try: + moment = timezone.parse(self.target_time) + except ValueError as e: + raise ValueError( + f"start_from_trigger=True requires a static target_time, not a template: {self.target_time!r}" + ) from e self.start_trigger_args.trigger_kwargs = dict( - moment=timezone.parse(self.target_time), + moment=moment, end_from_trigger=self.end_from_trigger, ) diff --git a/providers/standard/tests/unit/standard/sensors/test_date_time.py b/providers/standard/tests/unit/standard/sensors/test_date_time.py index 6b07eae4948ec..163af8ac5f4a8 100644 --- a/providers/standard/tests/unit/standard/sensors/test_date_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_date_time.py @@ -24,7 +24,7 @@ from airflow import macros from airflow.models.dag import DAG -from airflow.providers.standard.sensors.date_time import DateTimeSensor +from airflow.providers.standard.sensors.date_time import DateTimeSensor, DateTimeSensorAsync from tests_common.test_utils.version_compat import timezone @@ -124,3 +124,38 @@ def test_moment(self, native, target_time, expected_type): sensor.render_template_fields(ctx) assert isinstance(sensor._moment, expected_type) + + +class TestDateTimeSensorAsync: + @pytest.mark.parametrize( + "target_time", + [ + pendulum.datetime(2020, 7, 6, 13, tz="UTC"), + "2020-07-06T13:00:00+00:00", + ], + ) + def test_start_from_trigger_with_static_target_time(self, target_time): + with DAG( + dag_id="test_static_target_time", + schedule=None, + start_date=pendulum.datetime(2020, 1, 1, tz="UTC"), + ): + op = DateTimeSensorAsync(task_id="test", target_time=target_time, start_from_trigger=True) + + assert op.start_trigger_args.trigger_kwargs == { + "moment": pendulum.datetime(2020, 7, 6, 13, tz="UTC"), + "end_from_trigger": False, + } + + def test_start_from_trigger_with_templated_target_time_raises(self): + with DAG( + dag_id="test_templated_target_time", + schedule=None, + start_date=pendulum.datetime(2020, 1, 1, tz="UTC"), + ): + with pytest.raises(ValueError, match="requires a static target_time"): + DateTimeSensorAsync( + task_id="test", + target_time="{{ data_interval_end.tomorrow().replace(hour=1) }}", + start_from_trigger=True, + )