Fix DateTimeSensorAsync crashing Dag parsing with templated target_time

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,
             )
 

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,
+                )