[GitHub] [airflow] ashb commented on a diff in pull request #23153: Fix TaskInstance actions with upstream/downstream

[GitBox <gi...@apache.org>]

ashb commented on code in PR #23153:
URL: https://github.com/apache/airflow/pull/23153#discussion_r855453990


##########
airflow/models/taskinstance.py:
##########
@@ -2529,6 +2536,28 @@ def filter_for_tis(tis: Iterable[Union["TaskInstance", TaskInstanceKey]]) -> Opt
             (ti.key.primary for ti in tis),
         )
 
+    @classmethod
+    def filter_for_task_id_map_index_lists(cls, vals):
+        """
+        Build an SQLAlchemy filter for a list where each element can contain
+        whether a task_id, or a tuple of (task_id,map_index)
+
+        :meta private:
+        """
+        # Compute a filter for TI.task_id and TI.map_index based on input values
+        # For each item, it will either be a task_id, or (task_id, map_index)
+        task_id_only = list(filter(lambda v: isinstance(v, str), vals))
+        with_map_index = list(filter(lambda v: not isinstance(v, str), vals))
+        filters = []
+
+        if task_id_only:
+            filters.append(cls.task_id.in_(task_id_only))
+        if with_map_index:
+            filters.append(
+                tuple_in_condition((cls.task_id, cls.map_index), with_map_index),
+            )
+        return or_(*filters) if len(filters) > 1 else filters[0]

Review Comment:
   In my head this function should never result in 0 filters. I should assert that



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org