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..cc76eaa1fc9d9 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/date_time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/date_time.py @@ -21,6 +21,8 @@ from collections.abc import Sequence from typing import TYPE_CHECKING, Any, NoReturn +from pendulum.parsing.exceptions import ParserError + from airflow.providers.common.compat.sdk import BaseSensorOperator, timezone from airflow.providers.standard.triggers.temporal import DateTimeTrigger from airflow.providers.standard.version_compat import AIRFLOW_V_3_0_PLUS @@ -125,8 +127,18 @@ def __init__( self.start_from_trigger = start_from_trigger if self.start_from_trigger: + try: + moment = timezone.parse(self.target_time) + except ParserError as e: + raise ValueError( + "DateTimeSensorAsync does not support start_from_trigger=True with a templated " + f"target_time (got {self.target_time!r}). The trigger arguments are resolved once " + "at Dag-parse time, before template rendering happens, so a Jinja template cannot " + "be evaluated yet. Use a static datetime/ISO string for target_time, or set " + "start_from_trigger=False and use deferrable=True instead." + ) 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..557d6aad247f9 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,35 @@ def test_moment(self, native, target_time, expected_type): sensor.render_template_fields(ctx) assert isinstance(sensor._moment, expected_type) + + +class TestDateTimeSensorAsync: + @classmethod + def setup_class(cls): + args = {"owner": "airflow", "start_date": DEFAULT_DATE} + cls.dag = DAG("test_dag_async", schedule=None, default_args=args) + + def test_start_from_trigger_with_static_target_time(self): + """A static, already-resolvable target_time should keep working with start_from_trigger=True.""" + op = DateTimeSensorAsync( + task_id="static_target_time", + target_time="2020-07-06T13:00:00+00:00", + start_from_trigger=True, + dag=self.dag, + ) + assert op.start_trigger_args.trigger_kwargs["moment"] == pendulum.datetime(2020, 7, 6, 13, tz="UTC") + + def test_start_from_trigger_with_templated_target_time_raises_value_error(self): + """ + start_from_trigger=True resolves trigger_kwargs once, at Dag-parse time, before template + rendering happens. A templated target_time can't be resolved yet at that point, so this + must raise a clear ValueError instead of letting the Jinja string reach ``pendulum.parse`` + and crash the entire Dag-file parse with an opaque ParserError. + """ + with pytest.raises(ValueError, match="does not support start_from_trigger=True"): + DateTimeSensorAsync( + task_id="templated_target_time", + target_time="{{ data_interval_end.tomorrow().replace(hour=1) }}", + start_from_trigger=True, + dag=self.dag, + )