uranusjr commented on code in PR #27491:
URL: https://github.com/apache/airflow/pull/27491#discussion_r1024805382
##########
airflow/models/dagrun.py:
##########
@@ -730,30 +729,47 @@ def _get_ready_tis(
finished_tis=finished_tis,
)
+ def _expand_mapped_task_if_needed(ti: TI) -> Iterable[TI] | None:
+ """Try to expand the ti, if needed.
+
+ If the ti needs expansion, newly created task instances are
+ returned. The original ti is modified in-place and assigned the
+ ``map_index`` of 0.
+
+ If the ti does not need expansion, either because the task is not
+ mapped, or has already been expanded, *None* is returned.
+ """
+ if ti.map_index >= 0: # Already expanded, we're good.
+ return None
+ try:
+ expanded_tis, _ = ti.task.expand_mapped_task(self.run_id,
session=session)
+ except NotMapped: # Not a mapped task, nothing needed.
+ return None
+ if expanded_tis:
+ assert expanded_tis[0] is ti
Review Comment:
I think it’s more just for readability. As the function is implemented,
`expanded_tis[0]` is always the same instance as the original ti before it’s
expanded, this simply reminds the reader of this fact.
--
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]
