SameerMesiah97 commented on code in PR #70576:
URL: https://github.com/apache/airflow/pull/70576#discussion_r3695702402
##########
providers/standard/src/airflow/providers/standard/triggers/temporal.py:
##########
@@ -37,24 +37,77 @@ class DateTimeTrigger(BaseTrigger):
The provided datetime MUST be in UTC.
:param moment: when to yield event
+ :param target_time: a Jinja-templated ``target_time`` string that has not
yet been rendered.
+ Used instead of ``moment`` only when this trigger is created via
``start_from_trigger`` with
+ a ``target_time`` that could not be resolved to a concrete datetime at
Dag-parse time (i.e.
+ it is a template).
+ Mutually exclusive with ``moment``. ``target_time`` is one of
``DateTimeSensor``'s
+ ``template_fields``, so the triggerer renders it in place (see
+ ``BaseTrigger.render_template_fields``) before ``run()`` is invoked;
it is then parsed into
+ ``moment`` on first use. This mirrors how ``FileSensor`` defers
rendering of a templated
+ ``filepath`` to the triggerer.
:param end_from_trigger: whether the trigger should mark the task
successful after time condition
reached or resume the task after time condition reached.
"""
- def __init__(self, moment: datetime.datetime, *, end_from_trigger: bool =
False) -> None:
+ def __init__(
+ self,
+ moment: datetime.datetime | None = None,
+ *,
+ target_time: str | None = None,
+ end_from_trigger: bool = False,
+ ) -> None:
super().__init__()
+ if moment is None and target_time is None:
+ raise TypeError("DateTimeTrigger requires either 'moment' or
'target_time' to be set")
+ if moment is not None and target_time is not None:
+ raise TypeError("DateTimeTrigger accepts only one of 'moment' or
'target_time', not both")
+
+ self.end_from_trigger = end_from_trigger
+ # Kept around (and, when set, treated as a template field, see
task_instance.setter in
+ # BaseTrigger) so an unrendered `target_time` can be handed to the
triggerer and rendered
+ # there before `run()`/`serialize()` need a concrete moment.
+ self.target_time = target_time
+
+ self.moment: pendulum.DateTime | None
+ if moment is None:
+ self.moment = None
+ return
if not isinstance(moment, datetime.datetime):
raise TypeError(f"Expected datetime.datetime type for moment. Got
{type(moment)}")
# Make sure it's in UTC
if moment.tzinfo is None:
raise ValueError("You cannot pass naive datetimes")
- self.moment: pendulum.DateTime = timezone.convert_to_utc(moment)
- self.end_from_trigger = end_from_trigger
+ self.moment = timezone.convert_to_utc(moment)
+
+ def _resolve_moment(self) -> pendulum.DateTime:
+ """Return ``moment``, parsing it from a (by now rendered)
``target_time`` if needed."""
+ if self.moment is not None:
+ return self.moment
+ if not self.target_time:
+ raise TypeError("DateTimeTrigger requires either 'moment' or
'target_time' to be set")
+ try:
+ parsed = timezone.parse(self.target_time)
+ except ValueError as e:
+ raise ValueError(
+ f"Could not parse target_time {self.target_time!r} as a
datetime after template "
+ "rendering. start_from_trigger requires target_time to render
to a static datetime "
+ "or ISO-8601 string."
+ ) from e
+ self.moment = timezone.convert_to_utc(parsed)
+ return self.moment
def serialize(self) -> tuple[str, dict[str, Any]]:
+ # `target_time` is deliberately not round-tripped here: it is folded
into a concrete
+ # `moment` by `_resolve_moment()` before serialization, the same way
`TimeDeltaTrigger`
+ # folds `delta` into `moment` (see the exclusion for this class in
+ # scripts/ci/prek/check_trigger_serialize_init.py). By the time
serialize() is called,
+ # target_time has either already been rendered by the triggerer (see
+ # BaseTrigger.render_template_fields) or resolution raises a clear
error -- there is no
+ # unrendered template left to preserve.
Review Comment:
Not sure if this comment is needed. The code is self-explanatory.
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -115,10 +119,44 @@ def __init__(
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- self.start_trigger_args.trigger_kwargs = dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
- )
+ try:
+ moment = self._moment
+ except ValueError:
+ # target_time couldn't be parsed as a static datetime at
Dag-parse time. This is
+ # the normal, documented case of target_time being a Jinja
template, e.g.
+ # "{{ data_interval_end.tomorrow().replace(hour=1) }}" -- not
necessarily bad input.
+ moment = None
+
+ if moment is not None:
+ self.start_trigger_args.trigger_kwargs = dict(
+ moment=moment,
+ end_from_trigger=self.end_from_trigger,
+ )
+ elif AIRFLOW_V_3_3_PLUS:
+ # Hand the raw template string to the trigger under the same
name as this
+ # operator's template field ("target_time"). The triggerer
renders any
+ # start_trigger_args kwarg whose name matches an operator
template field before
+ # running the trigger (see BaseTrigger.task_instance /
render_template_fields),
+ # the same mechanism FileSensor relies on for a templated
`filepath`.
+ self.start_trigger_args.trigger_kwargs = dict(
+ target_time=self.target_time,
+ end_from_trigger=self.end_from_trigger,
+ )
+ else:
+ # Airflow < 3.3 triggerers can't render template fields on a
trigger before it
+ # runs, so a templated target_time can never be resolved via
start_from_trigger.
+ # Falling back to the normal worker-deferred path (execute())
avoids crashing Dag
+ # parsing on every parse cycle; execute() runs after the
scheduler/worker has
+ # already rendered target_time normally.
Review Comment:
Is this comment needed? Looks like the warning message communicates the
necessary information. I think this comment can be removed or trimmed.
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -88,6 +88,10 @@ 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 either a static ``target_time`` (a datetime or ISO-8601
string) or, on
+ Airflow >= 3.3, a templated ``target_time`` that the triggerer can
render before the trigger
+ runs. On earlier Airflow versions a templated ``target_time`` cannot
be resolved this way, so
+ ``start_from_trigger`` is disabled with a warning and the task defers
from the worker instead.
Review Comment:
This can be more concise. Please see the below:
```
:param start_from_trigger: Start the task directly from the triggerer
instead of a worker.
Supports static ``target_time`` values and, on Airflow >= 3.3, templated
``target_time`` values.
```
##########
providers/standard/src/airflow/providers/standard/triggers/temporal.py:
##########
@@ -37,24 +37,77 @@ class DateTimeTrigger(BaseTrigger):
The provided datetime MUST be in UTC.
:param moment: when to yield event
+ :param target_time: a Jinja-templated ``target_time`` string that has not
yet been rendered.
+ Used instead of ``moment`` only when this trigger is created via
``start_from_trigger`` with
+ a ``target_time`` that could not be resolved to a concrete datetime at
Dag-parse time (i.e.
+ it is a template).
+ Mutually exclusive with ``moment``. ``target_time`` is one of
``DateTimeSensor``'s
+ ``template_fields``, so the triggerer renders it in place (see
+ ``BaseTrigger.render_template_fields``) before ``run()`` is invoked;
it is then parsed into
+ ``moment`` on first use. This mirrors how ``FileSensor`` defers
rendering of a templated
+ ``filepath`` to the triggerer.
:param end_from_trigger: whether the trigger should mark the task
successful after time condition
reached or resume the task after time condition reached.
"""
- def __init__(self, moment: datetime.datetime, *, end_from_trigger: bool =
False) -> None:
+ def __init__(
+ self,
+ moment: datetime.datetime | None = None,
+ *,
+ target_time: str | None = None,
+ end_from_trigger: bool = False,
+ ) -> None:
super().__init__()
+ if moment is None and target_time is None:
+ raise TypeError("DateTimeTrigger requires either 'moment' or
'target_time' to be set")
+ if moment is not None and target_time is not None:
+ raise TypeError("DateTimeTrigger accepts only one of 'moment' or
'target_time', not both")
+
+ self.end_from_trigger = end_from_trigger
+ # Kept around (and, when set, treated as a template field, see
task_instance.setter in
+ # BaseTrigger) so an unrendered `target_time` can be handed to the
triggerer and rendered
+ # there before `run()`/`serialize()` need a concrete moment.
+ self.target_time = target_time
+
+ self.moment: pendulum.DateTime | None
+ if moment is None:
+ self.moment = None
+ return
if not isinstance(moment, datetime.datetime):
raise TypeError(f"Expected datetime.datetime type for moment. Got
{type(moment)}")
# Make sure it's in UTC
if moment.tzinfo is None:
raise ValueError("You cannot pass naive datetimes")
- self.moment: pendulum.DateTime = timezone.convert_to_utc(moment)
- self.end_from_trigger = end_from_trigger
+ self.moment = timezone.convert_to_utc(moment)
+
+ def _resolve_moment(self) -> pendulum.DateTime:
+ """Return ``moment``, parsing it from a (by now rendered)
``target_time`` if needed."""
+ if self.moment is not None:
+ return self.moment
+ if not self.target_time:
+ raise TypeError("DateTimeTrigger requires either 'moment' or
'target_time' to be set")
Review Comment:
Why duplicate the validation logic in the constructor here? Will it not be
guaranteed that `moment` or `target_time` is set by the time `_resolve_moment`
is called?
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -115,10 +119,44 @@ def __init__(
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- self.start_trigger_args.trigger_kwargs = dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
- )
+ try:
+ moment = self._moment
+ except ValueError:
+ # target_time couldn't be parsed as a static datetime at
Dag-parse time. This is
+ # the normal, documented case of target_time being a Jinja
template, e.g.
+ # "{{ data_interval_end.tomorrow().replace(hour=1) }}" -- not
necessarily bad input.
+ moment = None
+
+ if moment is not None:
+ self.start_trigger_args.trigger_kwargs = dict(
+ moment=moment,
+ end_from_trigger=self.end_from_trigger,
+ )
+ elif AIRFLOW_V_3_3_PLUS:
+ # Hand the raw template string to the trigger under the same
name as this
+ # operator's template field ("target_time"). The triggerer
renders any
+ # start_trigger_args kwarg whose name matches an operator
template field before
+ # running the trigger (see BaseTrigger.task_instance /
render_template_fields),
+ # the same mechanism FileSensor relies on for a templated
`filepath`.
Review Comment:
The comment is too long. Please see the below;
```
# Pass the unresolved template to the trigger. On Airflow >= 3.3 the
# triggerer renders trigger kwargs that correspond to template fields before
# starting the trigger.
```
##########
providers/standard/src/airflow/providers/standard/triggers/temporal.py:
##########
@@ -37,24 +37,77 @@ class DateTimeTrigger(BaseTrigger):
The provided datetime MUST be in UTC.
:param moment: when to yield event
+ :param target_time: a Jinja-templated ``target_time`` string that has not
yet been rendered.
+ Used instead of ``moment`` only when this trigger is created via
``start_from_trigger`` with
+ a ``target_time`` that could not be resolved to a concrete datetime at
Dag-parse time (i.e.
+ it is a template).
+ Mutually exclusive with ``moment``. ``target_time`` is one of
``DateTimeSensor``'s
+ ``template_fields``, so the triggerer renders it in place (see
+ ``BaseTrigger.render_template_fields``) before ``run()`` is invoked;
it is then parsed into
+ ``moment`` on first use. This mirrors how ``FileSensor`` defers
rendering of a templated
+ ``filepath`` to the triggerer.
Review Comment:
This is too long as well. Please see the below:
```
:param target_time: Templated ``target_time`` value to resolve when the
trigger
runs. Used instead of ``moment`` when the target time cannot be
determined
during operator initialization. Mutually exclusive with ``moment``.
```
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -115,10 +119,44 @@ def __init__(
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- self.start_trigger_args.trigger_kwargs = dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
- )
+ try:
+ moment = self._moment
+ except ValueError:
+ # target_time couldn't be parsed as a static datetime at
Dag-parse time. This is
+ # the normal, documented case of target_time being a Jinja
template, e.g.
+ # "{{ data_interval_end.tomorrow().replace(hour=1) }}" -- not
necessarily bad input.
+ moment = None
Review Comment:
This catches all `ValueErrors `and treats them as if the value were simply
"not yet renderable". Is that intentional? For example, an invalid static value
like "not-a-date" would now follow the same path as an unresolved Jinja
template. Should those cases be distinguished so genuinely invalid input still
fails eagerly?
##########
providers/standard/tests/unit/standard/sensors/test_date_time.py:
##########
@@ -157,3 +157,54 @@ def
test_async_start_from_trigger_localizes_naive_datetime(self):
dag=self.dag,
)
assert op.start_trigger_args.trigger_kwargs["moment"] ==
pendulum.datetime(2020, 1, 1, tz="UTC")
+
+ @patch("airflow.providers.standard.sensors.date_time.AIRFLOW_V_3_3_PLUS",
True)
+ def test_async_start_from_trigger_templated_target_time_on_3_3_plus(self):
+ """
+ On Airflow >= 3.3 a templated target_time must not crash Dag parsing
(#70284): the raw
+ template is handed to the trigger under the "target_time" key
(matching the operator's
+ template field) so the triggerer can render it before the trigger runs.
+ """
+ op = DateTimeSensorAsync(
+ task_id="async_templated_3_3",
+ target_time="{{ data_interval_end.tomorrow().replace(hour=1) }}",
+ start_from_trigger=True,
+ dag=self.dag,
+ )
+ assert op.start_from_trigger is True
+ assert op.start_trigger_args.trigger_kwargs == {
+ "target_time": "{{ data_interval_end.tomorrow().replace(hour=1)
}}",
+ "end_from_trigger": False,
+ }
+ # No moment key should be present -- moment can't be computed until
the template renders.
+ assert "moment" not in op.start_trigger_args.trigger_kwargs
+
+ @patch("airflow.providers.standard.sensors.date_time.AIRFLOW_V_3_3_PLUS",
False)
+ def test_async_start_from_trigger_templated_target_time_pre_3_3(self,
capsys):
+ """
+ Before Airflow 3.3, the triggerer can't render template fields on a
trigger before it
+ runs, so a templated target_time can never resolve via
start_from_trigger. Rather than
+ crashing Dag parsing, fall back to the normal worker-deferred path.
+ """
+ op = DateTimeSensorAsync(
+ task_id="async_templated_pre_3_3",
+ target_time="{{ data_interval_end.tomorrow().replace(hour=1) }}",
+ start_from_trigger=True,
+ dag=self.dag,
+ )
+ assert op.start_from_trigger is False
+ # Airflow's task logger uses structlog and writes straight to stdout,
so it isn't
+ # captured by the stdlib-logging-based `caplog` fixture; assert on
stdout instead.
+ assert "requires a static target_time" in capsys.readouterr().out
+
+ @patch("airflow.providers.standard.sensors.date_time.AIRFLOW_V_3_3_PLUS",
False)
+ def
test_async_start_from_trigger_static_target_time_pre_3_3_unaffected(self):
+ """A static target_time must keep working identically regardless of
AIRFLOW_V_3_3_PLUS."""
+ op = DateTimeSensorAsync(
+ task_id="async_static_pre_3_3",
+ target_time="2020-01-01T00:00:00+00:00",
+ start_from_trigger=True,
+ dag=self.dag,
+ )
+ assert op.start_from_trigger is True
+ assert op.start_trigger_args.trigger_kwargs["moment"] ==
pendulum.parse("2020-01-01T00:00:00+00:00")
Review Comment:
Would it be worth adding a test for a non-templated bt invalid `target_time`
(e.g. "not-a-date") so the expected behaviour for genuinely invalid input is
explicit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
