fix(sdk): Allow non-pythonic names for graph components' task's outpu…

…ts. Fixes #4514. (#4515)

* add tests for pythonic and non-pythonic component outputs

* fix: graph for non-pythonic container output's names

Loading container component from component.yaml creates both
pythonic and original output names. Graph component iterated over
all outputs, using pythonic-to-output conversion on all. If some
of the names are not identical to their pythonic versions, they
rised KeyError on the lookup table.

This commit fixes this problem by using default value for the lookup.

* remove depythonification of outputs - not needed anymore

sdk/python/kfp/components/_components.py

@@ -531,13 +531,10 @@ def resolve_argument(argument):
         task_component_spec = task_factory.component_spec
 
         input_name_to_pythonic = generate_unique_name_conversion_table([input.name for input in task_component_spec.inputs or []], _sanitize_python_function_name)
-        output_name_to_pythonic = generate_unique_name_conversion_table([output.name for output in task_component_spec.outputs or []], _sanitize_python_function_name)
-        pythonic_output_name_to_original = {pythonic_name: original_name for original_name, pythonic_name in output_name_to_pythonic.items()}
         pythonic_task_arguments = {input_name_to_pythonic[input_name]: argument for input_name, argument in task_arguments.items()}
 
         task_obj = task_factory(**pythonic_task_arguments)
-        task_outputs_with_pythonic_names = task_obj.outputs
-        task_outputs_with_original_names = {pythonic_output_name_to_original[pythonic_output_name]: output_value for pythonic_output_name, output_value in task_outputs_with_pythonic_names.items()}
+        task_outputs_with_original_names = {output.name: task_obj.outputs[output.name] for output in task_component_spec.outputs or []}
         outputs_of_tasks[task_id] = task_outputs_with_original_names
 
     resolved_graph_outputs = OrderedDict([(output_name, resolve_argument(argument)) for output_name, argument in graph.output_values.items()])

sdk/python/tests/dsl/component_bridge_tests.py

@@ -266,3 +266,78 @@ def test_prevent_passing_container_op_as_argument(self):
             component(input_1=task1, input_2="value 2")
         with self.assertRaises(TypeError):
             component(input_1="value 1", input_2=task1)
+
+    def test_pythonic_container_output_handled_by_graph(self):
+        component_a = textwrap.dedent('''\
+          inputs: []
+          outputs:
+            - {name: out1, type: str}
+          implementation:
+            graph:
+              tasks:
+                some-container:
+                  arguments: {}
+                  componentRef:
+                    spec:
+                      outputs:
+                      - {name: out1, type: str}
+                      implementation:
+                        container:
+                          image: busybox
+                          command: [bash, -c, 'mkdir -p "$(dirname "$0")"; date > "$0"', {outputPath: out1}]
+              outputValues:
+                out1:
+                  taskOutput:
+                    taskId: some-container
+                    outputName: out1
+        ''')
+        component_b = textwrap.dedent('''\
+            inputs:
+            - {name: in1, type: str}
+            implementation:
+              container:
+                image: busybox
+                command: [echo, {inputValue: in1}]
+        '''
+                                      )
+        task_factory_a = load_component_from_text(component_a)
+        task_factory_b = load_component_from_text(component_b)
+        a_task = task_factory_a()
+        b_task = task_factory_b(in1=a_task.outputs['out1'])
+
+    def test_nonpythonic_container_output_handled_by_graph(self):
+        component_a = textwrap.dedent('''\
+          inputs: []
+          outputs:
+            - {name: out1, type: str}
+          implementation:
+            graph:
+              tasks:
+                some-container:
+                  arguments: {}
+                  componentRef:
+                    spec:
+                      outputs:
+                      - {name: out-1, type: str}
+                      implementation:
+                        container:
+                          image: busybox
+                          command: [bash, -c, 'mkdir -p "$(dirname "$0")"; date > "$0"', {outputPath: out-1}]
+              outputValues:
+                out1:
+                  taskOutput:
+                    taskId: some-container
+                    outputName: out-1
+        ''')
+        component_b = textwrap.dedent('''\
+            inputs:
+            - {name: in1, type: str}
+            implementation:
+              container:
+                image: busybox
+                command: [echo, {inputValue: in1}]
+        ''')
+        task_factory_a = load_component_from_text(component_a)
+        task_factory_b = load_component_from_text(component_b)
+        a_task = task_factory_a()
+        b_task = task_factory_b(in1=a_task.outputs['out1'])