Lee-W commented on code in PR #70576:
URL: https://github.com/apache/airflow/pull/70576#discussion_r3965858459
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -116,16 +118,49 @@ def __init__(
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- # Replaced rather than mutated: ``start_trigger_args`` is a class
attribute, so
- # assigning through it would overwrite the arguments of every
other task built
- # from this operator.
- self.start_trigger_args = dataclasses.replace(
- self.start_trigger_args,
- trigger_kwargs=dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
- ),
- )
+ try:
+ moment = self._moment
+ except ValueError:
+ if not self._looks_like_template(self.target_time):
+ # genuinely invalid input (e.g. "not-a-date"), not a
template: fail fast
+ raise
+ moment = None
+
+ # start_trigger_args is a class attribute, so replace it rather
than mutate it
+ if moment is not None:
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args,
+ trigger_kwargs=dict(
+ moment=moment,
+ end_from_trigger=self.end_from_trigger,
+ ),
+ )
+ elif AIRFLOW_V_3_3_PLUS:
+ # 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.
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args,
+ trigger_kwargs=dict(
+ target_time=self.target_time,
+ end_from_trigger=self.end_from_trigger,
+ ),
+ )
+ else:
+ self.log.warning(
+ "start_from_trigger=True requires a static target_time on
Airflow < 3.3, but "
+ "%r looks like a template for task %r. Disabling
start_from_trigger and "
+ "deferring from the worker instead. Upgrade to Airflow >=
3.3 to defer "
+ "directly from the triggerer with a templated
target_time.",
+ self.target_time,
+ self.task_id,
+ )
+ self.start_from_trigger = False
+
+ @staticmethod
+ def _looks_like_template(target_time: Any) -> bool:
+ """Whether ``target_time`` contains unrendered Jinja delimiters."""
+ return isinstance(target_time, str) and ("{{" in target_time or "{%"
in target_time)
Review Comment:
let's also check }} and %}
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -116,16 +118,49 @@ def __init__(
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- # Replaced rather than mutated: ``start_trigger_args`` is a class
attribute, so
- # assigning through it would overwrite the arguments of every
other task built
- # from this operator.
- self.start_trigger_args = dataclasses.replace(
- self.start_trigger_args,
- trigger_kwargs=dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
- ),
- )
+ try:
+ moment = self._moment
+ except ValueError:
+ if not self._looks_like_template(self.target_time):
+ # genuinely invalid input (e.g. "not-a-date"), not a
template: fail fast
Review Comment:
```suggestion
```
##########
providers/standard/src/airflow/providers/standard/sensors/date_time.py:
##########
@@ -116,16 +118,49 @@ def __init__(
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- # Replaced rather than mutated: ``start_trigger_args`` is a class
attribute, so
- # assigning through it would overwrite the arguments of every
other task built
- # from this operator.
- self.start_trigger_args = dataclasses.replace(
- self.start_trigger_args,
- trigger_kwargs=dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
- ),
- )
+ try:
+ moment = self._moment
+ except ValueError:
+ if not self._looks_like_template(self.target_time):
+ # genuinely invalid input (e.g. "not-a-date"), not a
template: fail fast
Review Comment:
I don't think we need this comment
--
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]
