Lee-W commented on code in PR #39912:
URL: https://github.com/apache/airflow/pull/39912#discussion_r1684623009
##########
airflow/models/mappedoperator.py:
##########
@@ -729,6 +731,67 @@ def _get_unmap_kwargs(self, mapped_kwargs: Mapping[str,
Any], *, strict: bool) -
"params": params,
}
+ def _expand_start_from_trigger(self, *, context: Context, session:
Session) -> bool:
+ """
+ Get the kwargs to create the unmapped start_from_trigger.
+
+ This method is for allowing mapped operator to start execution from
triggerer.
+ """
+ if not self.start_trigger_args:
+ return False
+
+ mapped_kwargs, _ = self._expand_mapped_kwargs(context, session,
include_xcom=False)
+ if self._disallow_kwargs_override:
+ prevent_duplicates(
+ self.partial_kwargs,
+ mapped_kwargs,
+ fail_reason="unmappable or already specified",
+ )
+
+ # Ordering is significant; mapped kwargs should override partial ones.
+ return mapped_kwargs.get(
+ "start_from_trigger",
self.partial_kwargs.get("start_from_trigger", self.start_from_trigger)
+ )
+
+ def _expand_start_trigger_args(self, *, context: Context, session:
Session) -> StartTriggerArgs:
+ """
+ Get the kwargs to create the unmapped start_trigger_args.
+
+ This method is for allowing mapped operator to start execution from
triggerer.
+ """
+ if not self.start_trigger_args:
+ raise AirflowException(
Review Comment:
It actually wouldn't happen (ideally). Just try to make mypy happy. But I
changed it to TaskDeferralError in `_defer_task`, which seems to make more
sense after the AbstractOperator.expand_... refactoring
--
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]
