Fix executor handling for collected list artifacts (#12707)

* Fix executor handling for collected list artifacts

Signed-off-by: sduvvuri1603 <[email protected]>

* Format executor loop per yapf

Signed-off-by: sduvvuri1603 <[email protected]>

* chore: normalize executor string quotes

Signed-off-by: sduvvuri1603 <[email protected]>

* refactor: dedupe list artifact resolver

Signed-off-by: sduvvuri1603 <[email protected]>
Co-authored-by: Cursor <[email protected]>

* chore: docformat executor helper docstring

Signed-off-by: sduvvuri1603 <[email protected]>
Co-authored-by: Cursor <[email protected]>

---------

Signed-off-by: sduvvuri1603 <[email protected]>
Co-authored-by: Cursor <[email protected]>

Author: sduvvuri1603

sdk/python/kfp/dsl/executor.py

@@ -71,10 +71,8 @@ def assign_input_and_output_artifacts(self) -> None:
                     type_annotations.is_list_of_artifacts(annotation.__origin__)
                 ) or type_annotations.is_list_of_artifacts(annotation)
                 if is_list_of_artifacts:
-                    # Get the annotation of the inner type of the list
-                    # to use when creating the artifacts
-                    inner_annotation = type_annotations.get_inner_type(
-                        annotation)
+                    inner_annotation = self._resolve_list_artifact_inner_type(
+                        input_name=name, list_annotation=annotation)
 
                     self.input_artifacts[name] = [
                         self.make_artifact(
@@ -104,6 +102,41 @@ def assign_input_and_output_artifacts(self) -> None:
                 self.output_artifacts[name] = output_artifact
                 makedirs_recursively(output_artifact.path)
 
+    def _resolve_list_artifact_inner_type(self, input_name: str,
+                                          list_annotation: Any) -> Any:
+        """Unwraps nested list annotations and validates the inner artifact
+        type."""
+        inner_annotation = type_annotations.get_inner_type(list_annotation)
+
+        if inner_annotation is None:
+            raise TypeError(
+                f"Input '{input_name}' expects a list of artifacts, but "
+                f'received {list_annotation!r} without an inner type.')
+
+        while type_annotations.is_list_of_artifacts(inner_annotation):
+            if isinstance(inner_annotation, tuple):
+                raise TypeError(
+                    f"Input '{input_name}' expects a single artifact type but "
+                    f'received a union {inner_annotation!r}.')
+            inner_annotation = type_annotations.get_inner_type(inner_annotation)
+            if inner_annotation is None:
+                raise TypeError(
+                    f"Input '{input_name}' expects a list of artifacts, but "
+                    f'could not determine the inner annotation from '
+                    f'{list_annotation!r}.')
+
+        if isinstance(inner_annotation, tuple):
+            raise TypeError(
+                f"Input '{input_name}' expects a single artifact type but "
+                f'received a union {inner_annotation!r}.')
+        if not type_annotations.is_artifact_class(inner_annotation):
+            raise TypeError(
+                f"Input '{input_name}' expects a list of artifacts, but "
+                f'received {list_annotation!r} whose inner type '
+                f'{inner_annotation!r} is not an artifact.')
+
+        return inner_annotation
+
     def make_artifact(
         self,
         runtime_artifact: Dict,
@@ -113,6 +146,9 @@ def make_artifact(
     ) -> Any:
         annotation = func.__annotations__.get(
             name) if annotation is None else annotation
+        if type_annotations.is_list_of_artifacts(annotation):
+            annotation = self._resolve_list_artifact_inner_type(
+                input_name=name, list_annotation=annotation)
         if isinstance(annotation, type_annotations.InputPath):
             schema_title, _ = annotation.type.split('@')
             if schema_title in artifact_types._SCHEMA_TITLE_TO_TYPE:

sdk/python/kfp/dsl/executor_test.py

@@ -1263,6 +1263,55 @@ def test_func(input_artifact: Input[Artifact]):
 
         self.assertDictEqual(output_metadata, {})
 
+    def test_list_of_artifact_input(self):
+        executor_input = """\
+    {
+      "inputs": {
+        "artifacts": {
+          "input_datasets": {
+            "artifacts": [
+              {
+                "metadata": {"rows": 10},
+                "name": "datasets/0",
+                "type": {
+                  "schemaTitle": "system.Dataset"
+                },
+                "uri": "gs://some-bucket/output/input_dataset_0"
+              },
+              {
+                "metadata": {"rows": 20},
+                "name": "datasets/1",
+                "type": {
+                  "schemaTitle": "system.Dataset"
+                },
+                "uri": "gs://some-bucket/output/input_dataset_1"
+              }
+            ]
+          }
+        }
+      },
+      "outputs": {
+        "outputFile": "%(test_dir)s/output_metadata.json"
+      }
+    }
+    """
+
+        def test_func(input_datasets: Input[List[Dataset]]):
+            self.assertIsInstance(input_datasets, list)
+            self.assertLen(input_datasets, 2)
+            for index, artifact in enumerate(input_datasets):
+                self.assertIsInstance(artifact, Dataset)
+                self.assertEqual(artifact.name, f'datasets/{index}')
+                self.assertEqual(
+                    artifact.uri,
+                    f'gs://some-bucket/output/input_dataset_{index}')
+                self.assertEqual(artifact.metadata['rows'], (index + 1) * 10)
+
+        output_metadata = self.execute_and_load_output_metadata(
+            test_func, executor_input)
+
+        self.assertDictEqual(output_metadata, {})
+
     def test_single_artifact_input_pythonic(self):
         executor_input = """\
     {