We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
allow output artifact store configuration (vs hard coded)
it seems like the output artifacts are always stored in a specific minio service, port, namespace, bucket, secrets, etc (`minio-service.kubeflow:9000`). 

see: https://github.com/kubeflow/pipelines/blob/f40a22a3f4a8e06d20cf3e3f425b5058d5c87e0b/sdk/python/kfp/compiler/_op_to_template.py#L148

it would be great to make it flexible, e.g. allow using S3, or change namespace or bucket names.
i suggest making it configurable, i can do such PR if we agree its needed. 
flexible pipeline service (host) path in client SDK 
when creating an SDK `Client()` the path to `ml-pipeline` API service is loaded from a hard coded value (`ml-pipeline.kubeflow.svc.cluster.local:8888`) which indicate a specific k8s namespace. it can be valuable to load that default value from an env variable, i.e. changing the line in `_client.py` from:

`config.host = host if host else Client.IN_CLUSTER_DNS_NAME`

to:

`config.host = host or os.environ.get('ML_PIPELINE_DNS_NAME',Client.IN_CLUSTER_DNS_NAME)`

also note that when a user provide the `host` parameter, the ipython output points to the API server and not to the UI service (see the logic in `_get_url_prefix()`), it seems like a potential bug

if its acceptable i can submit a PR for the line change above
 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/kfp/components/_python_to_graph_component.py`
Content:
```
1 # Copyright 2019 Google LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 __all__ = [
16     'create_graph_component_from_pipeline_func',
17 ]
18 
19 
20 import inspect
21 from collections import OrderedDict
22 from typing import Callable
23 
24 from . import _components
25 from ._structures import TaskSpec, ComponentSpec, OutputSpec, GraphInputReference, TaskOutputArgument, GraphImplementation, GraphSpec
26 from ._naming import _make_name_unique_by_adding_index
27 from ._python_op import _extract_component_interface
28 from ._components import _create_task_factory_from_component_spec
29 
30 
31 def create_graph_component_from_pipeline_func(
32     pipeline_func: Callable,
33     output_component_file: str = None,
34     embed_component_specs: bool = False,
35 ) -> Callable:
36     '''Experimental! Creates graph component definition from a python pipeline function. The component file can be published for sharing.
37     Pipeline function is a function that only calls component functions and passes outputs to inputs.
38     This feature is experimental and lacks support for some of the DSL features like conditions and loops.
39     Only pipelines consisting of loaded components or python components are currently supported (no manually created ContainerOps or ResourceOps).
40 
41     Args:
42         pipeline_func: Python function to convert
43         output_component_file: Path of the file where the component definition will be written. The `component.yaml` file can then be published for sharing.
44         embed_component_specs: Whether to embed component definitions or just reference them. Embedding makes the graph component self-contained. Default is False.
45 
46     Returns:
47         A function representing the graph component. The component spec can be accessed using the .component_spec attribute.
48         The function will have the same parameters as the original function.
49         When called, the function will return a task object, corresponding to the graph component.
50         To reference the outputs of the task, use task.outputs["Output name"].
51 
52     Example:
53 
54         producer_op = load_component_from_file('producer/component.yaml')
55         processor_op = load_component_from_file('processor/component.yaml')
56 
57         def pipeline1(pipeline_param_1: int):
58             producer_task = producer_op()
59             processor_task = processor_op(pipeline_param_1, producer_task.outputs['Output 2'])
60 
61             return OrderedDict([
62                 ('Pipeline output 1', producer_task.outputs['Output 1']),
63                 ('Pipeline output 2', processor_task.outputs['Output 2']),
64             ])
65         
66         create_graph_component_from_pipeline_func(pipeline1, output_component_file='pipeline.component.yaml')
67     '''
68     component_spec = create_graph_component_spec_from_pipeline_func(pipeline_func, embed_component_specs)
69     if output_component_file:
70         from pathlib import Path
71         from ._yaml_utils import dump_yaml
72         component_dict = component_spec.to_dict()
73         component_yaml = dump_yaml(component_dict)
74         Path(output_component_file).write_text(component_yaml)
75 
76     return _create_task_factory_from_component_spec(component_spec)
77 
78 def create_graph_component_spec_from_pipeline_func(pipeline_func: Callable, embed_component_specs: bool = False) -> ComponentSpec:
79 
80     component_spec = _extract_component_interface(pipeline_func)
81     # Checking the function parameters - they should not have file passing annotations.
82     input_specs = component_spec.inputs or []
83     for input in input_specs:
84         if input._passing_style:
85             raise TypeError('Graph component function parameter "{}" cannot have file-passing annotation "{}".'.format(input.name, input._passing_style))
86 
87     task_map = OrderedDict() #Preserving task order
88 
89     from ._components import _create_task_spec_from_component_and_arguments
90     def task_construction_handler(
91         component_spec,
92         arguments,
93         component_ref,
94     ):
95         task = _create_task_spec_from_component_and_arguments(
96             component_spec=component_spec,
97             arguments=arguments,
98             component_ref=component_ref,
99         )
100 
101         #Rewriting task ids so that they're same every time
102         task_id = task.component_ref.spec.name or "Task"
103         task_id = _make_name_unique_by_adding_index(task_id, task_map.keys(), ' ')
104         for output_ref in task.outputs.values():
105             output_ref.task_output.task_id = task_id
106             output_ref.task_output.task = None
107         task_map[task_id] = task
108         # Remove the component spec from component reference unless it will make the reference empty or unless explicitly asked by the user
109         if not embed_component_specs and any([task.component_ref.name, task.component_ref.url, task.component_ref.digest]):
110             task.component_ref.spec = None
111 
112         return task #The handler is a transformation function, so it must pass the task through.
113 
114     # Preparing the pipeline_func arguments
115     # TODO: The key should be original parameter name if different
116     pipeline_func_args = {input.name: GraphInputReference(input_name=input.name).as_argument() for input in input_specs}
117 
118     try:
119         #Setting the handler to fix and catch the tasks.
120         # FIX: The handler only hooks container component creation
121         old_handler = _components._container_task_constructor
122         _components._container_task_constructor = task_construction_handler
123         
124         #Calling the pipeline_func with GraphInputArgument instances as arguments 
125         pipeline_func_result = pipeline_func(**pipeline_func_args)
126     finally:
127         _components._container_task_constructor = old_handler
128 
129 
130     # Getting graph outputs
131     output_names = [output.name for output in (component_spec.outputs or [])]
132 
133     if len(output_names) == 1 and output_names[0] == 'Output': # TODO: Check whether the NamedTuple syntax was used
134         pipeline_func_result = [pipeline_func_result]
135 
136     if isinstance(pipeline_func_result, tuple) and hasattr(pipeline_func_result, '_asdict'): # collections.namedtuple and typing.NamedTuple
137         pipeline_func_result = pipeline_func_result._asdict()
138 
139     if isinstance(pipeline_func_result, dict):
140         if output_names:
141             if set(output_names) != set(pipeline_func_result.keys()):
142                 raise ValueError('Returned outputs do not match outputs specified in the function signature: {} = {}'.format(str(set(pipeline_func_result.keys())), str(set(output_names))))
143 
144     if pipeline_func_result is None:
145         graph_output_value_map = {}
146     elif isinstance(pipeline_func_result, dict):
147         graph_output_value_map = OrderedDict(pipeline_func_result)
148     elif isinstance(pipeline_func_result, (list, tuple)):
149         if output_names:
150             if len(pipeline_func_result) != len(output_names):
151                 raise ValueError('Expected {} values from pipeline function, but got {}.'.format(len(output_names), len(pipeline_func_result)))
152             graph_output_value_map = OrderedDict((name_value[0], name_value[1]) for name_value in zip(output_names, pipeline_func_result))
153         else:
154             graph_output_value_map = OrderedDict((output_value.task_output.output_name, output_value) for output_value in pipeline_func_result) # TODO: Fix possible name non-uniqueness (e.g. use task id as prefix or add index to non-unique names)
155     else:
156         raise TypeError('Pipeline must return outputs as tuple or OrderedDict.')
157 
158     #Checking the pipeline_func output object types
159     for output_name, output_value in graph_output_value_map.items():
160         if not isinstance(output_value, TaskOutputArgument):
161             raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got "{output_name}" = "{}".'.format(output_name, str(output_value)))
162 
163     if not component_spec.outputs and graph_output_value_map:
164         component_spec.outputs = [OutputSpec(name=output_name, type=output_value.task_output.type) for output_name, output_value in graph_output_value_map.items()]
165 
166     component_spec.implementation = GraphImplementation(
167         graph=GraphSpec(
168             tasks=task_map,
169             output_values=graph_output_value_map,
170         )
171     )
172     return component_spec
173
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/kfp/components/_python_to_graph_component.py b/sdk/python/kfp/components/_python_to_graph_component.py
--- a/sdk/python/kfp/components/_python_to_graph_component.py
+++ b/sdk/python/kfp/components/_python_to_graph_component.py
@@ -158,7 +158,7 @@
     #Checking the pipeline_func output object types
     for output_name, output_value in graph_output_value_map.items():
         if not isinstance(output_value, TaskOutputArgument):
-            raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got "{output_name}" = "{}".'.format(output_name, str(output_value)))
+            raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got "{}" = "{}".'.format(output_name, str(output_value)))
 
     if not component_spec.outputs and graph_output_value_map:
         component_spec.outputs = [OutputSpec(name=output_name, type=output_value.task_output.type) for output_name, output_value in graph_output_value_map.items()]

{"golden_diff": "diff --git a/sdk/python/kfp/components/_python_to_graph_component.py b/sdk/python/kfp/components/_python_to_graph_component.py\n--- a/sdk/python/kfp/components/_python_to_graph_component.py\n+++ b/sdk/python/kfp/components/_python_to_graph_component.py\n@@ -158,7 +158,7 @@\n     #Checking the pipeline_func output object types\n     for output_name, output_value in graph_output_value_map.items():\n         if not isinstance(output_value, TaskOutputArgument):\n-            raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got \"{output_name}\" = \"{}\".'.format(output_name, str(output_value)))\n+            raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got \"{}\" = \"{}\".'.format(output_name, str(output_value)))\n \n     if not component_spec.outputs and graph_output_value_map:\n         component_spec.outputs = [OutputSpec(name=output_name, type=output_value.task_output.type) for output_name, output_value in graph_output_value_map.items()]\n", "issue": "allow output artifact store configuration (vs hard coded)\nit seems like the output artifacts are always stored in a specific minio service, port, namespace, bucket, secrets, etc (`minio-service.kubeflow:9000`). \r\n\r\nsee: https://github.com/kubeflow/pipelines/blob/f40a22a3f4a8e06d20cf3e3f425b5058d5c87e0b/sdk/python/kfp/compiler/_op_to_template.py#L148\r\n\r\nit would be great to make it flexible, e.g. allow using S3, or change namespace or bucket names.\r\ni suggest making it configurable, i can do such PR if we agree its needed. \nflexible pipeline service (host) path in client SDK \nwhen creating an SDK `Client()` the path to `ml-pipeline` API service is loaded from a hard coded value (`ml-pipeline.kubeflow.svc.cluster.local:8888`) which indicate a specific k8s namespace. it can be valuable to load that default value from an env variable, i.e. changing the line in `_client.py` from:\r\n\r\n`config.host = host if host else Client.IN_CLUSTER_DNS_NAME`\r\n\r\nto:\r\n\r\n`config.host = host or os.environ.get('ML_PIPELINE_DNS_NAME',Client.IN_CLUSTER_DNS_NAME)`\r\n\r\nalso note that when a user provide the `host` parameter, the ipython output points to the API server and not to the UI service (see the logic in `_get_url_prefix()`), it seems like a potential bug\r\n\r\nif its acceptable i can submit a PR for the line change above\r\n \n", "before_files": [{"content": "# Copyright 2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n__all__ = [\n    'create_graph_component_from_pipeline_func',\n]\n\n\nimport inspect\nfrom collections import OrderedDict\nfrom typing import Callable\n\nfrom . import _components\nfrom ._structures import TaskSpec, ComponentSpec, OutputSpec, GraphInputReference, TaskOutputArgument, GraphImplementation, GraphSpec\nfrom ._naming import _make_name_unique_by_adding_index\nfrom ._python_op import _extract_component_interface\nfrom ._components import _create_task_factory_from_component_spec\n\n\ndef create_graph_component_from_pipeline_func(\n    pipeline_func: Callable,\n    output_component_file: str = None,\n    embed_component_specs: bool = False,\n) -> Callable:\n    '''Experimental! Creates graph component definition from a python pipeline function. The component file can be published for sharing.\n    Pipeline function is a function that only calls component functions and passes outputs to inputs.\n    This feature is experimental and lacks support for some of the DSL features like conditions and loops.\n    Only pipelines consisting of loaded components or python components are currently supported (no manually created ContainerOps or ResourceOps).\n\n    Args:\n        pipeline_func: Python function to convert\n        output_component_file: Path of the file where the component definition will be written. The `component.yaml` file can then be published for sharing.\n        embed_component_specs: Whether to embed component definitions or just reference them. Embedding makes the graph component self-contained. Default is False.\n\n    Returns:\n        A function representing the graph component. The component spec can be accessed using the .component_spec attribute.\n        The function will have the same parameters as the original function.\n        When called, the function will return a task object, corresponding to the graph component.\n        To reference the outputs of the task, use task.outputs[\"Output name\"].\n\n    Example:\n\n        producer_op = load_component_from_file('producer/component.yaml')\n        processor_op = load_component_from_file('processor/component.yaml')\n\n        def pipeline1(pipeline_param_1: int):\n            producer_task = producer_op()\n            processor_task = processor_op(pipeline_param_1, producer_task.outputs['Output 2'])\n\n            return OrderedDict([\n                ('Pipeline output 1', producer_task.outputs['Output 1']),\n                ('Pipeline output 2', processor_task.outputs['Output 2']),\n            ])\n        \n        create_graph_component_from_pipeline_func(pipeline1, output_component_file='pipeline.component.yaml')\n    '''\n    component_spec = create_graph_component_spec_from_pipeline_func(pipeline_func, embed_component_specs)\n    if output_component_file:\n        from pathlib import Path\n        from ._yaml_utils import dump_yaml\n        component_dict = component_spec.to_dict()\n        component_yaml = dump_yaml(component_dict)\n        Path(output_component_file).write_text(component_yaml)\n\n    return _create_task_factory_from_component_spec(component_spec)\n\ndef create_graph_component_spec_from_pipeline_func(pipeline_func: Callable, embed_component_specs: bool = False) -> ComponentSpec:\n\n    component_spec = _extract_component_interface(pipeline_func)\n    # Checking the function parameters - they should not have file passing annotations.\n    input_specs = component_spec.inputs or []\n    for input in input_specs:\n        if input._passing_style:\n            raise TypeError('Graph component function parameter \"{}\" cannot have file-passing annotation \"{}\".'.format(input.name, input._passing_style))\n\n    task_map = OrderedDict() #Preserving task order\n\n    from ._components import _create_task_spec_from_component_and_arguments\n    def task_construction_handler(\n        component_spec,\n        arguments,\n        component_ref,\n    ):\n        task = _create_task_spec_from_component_and_arguments(\n            component_spec=component_spec,\n            arguments=arguments,\n            component_ref=component_ref,\n        )\n\n        #Rewriting task ids so that they're same every time\n        task_id = task.component_ref.spec.name or \"Task\"\n        task_id = _make_name_unique_by_adding_index(task_id, task_map.keys(), ' ')\n        for output_ref in task.outputs.values():\n            output_ref.task_output.task_id = task_id\n            output_ref.task_output.task = None\n        task_map[task_id] = task\n        # Remove the component spec from component reference unless it will make the reference empty or unless explicitly asked by the user\n        if not embed_component_specs and any([task.component_ref.name, task.component_ref.url, task.component_ref.digest]):\n            task.component_ref.spec = None\n\n        return task #The handler is a transformation function, so it must pass the task through.\n\n    # Preparing the pipeline_func arguments\n    # TODO: The key should be original parameter name if different\n    pipeline_func_args = {input.name: GraphInputReference(input_name=input.name).as_argument() for input in input_specs}\n\n    try:\n        #Setting the handler to fix and catch the tasks.\n        # FIX: The handler only hooks container component creation\n        old_handler = _components._container_task_constructor\n        _components._container_task_constructor = task_construction_handler\n        \n        #Calling the pipeline_func with GraphInputArgument instances as arguments \n        pipeline_func_result = pipeline_func(**pipeline_func_args)\n    finally:\n        _components._container_task_constructor = old_handler\n\n\n    # Getting graph outputs\n    output_names = [output.name for output in (component_spec.outputs or [])]\n\n    if len(output_names) == 1 and output_names[0] == 'Output': # TODO: Check whether the NamedTuple syntax was used\n        pipeline_func_result = [pipeline_func_result]\n\n    if isinstance(pipeline_func_result, tuple) and hasattr(pipeline_func_result, '_asdict'): # collections.namedtuple and typing.NamedTuple\n        pipeline_func_result = pipeline_func_result._asdict()\n\n    if isinstance(pipeline_func_result, dict):\n        if output_names:\n            if set(output_names) != set(pipeline_func_result.keys()):\n                raise ValueError('Returned outputs do not match outputs specified in the function signature: {} = {}'.format(str(set(pipeline_func_result.keys())), str(set(output_names))))\n\n    if pipeline_func_result is None:\n        graph_output_value_map = {}\n    elif isinstance(pipeline_func_result, dict):\n        graph_output_value_map = OrderedDict(pipeline_func_result)\n    elif isinstance(pipeline_func_result, (list, tuple)):\n        if output_names:\n            if len(pipeline_func_result) != len(output_names):\n                raise ValueError('Expected {} values from pipeline function, but got {}.'.format(len(output_names), len(pipeline_func_result)))\n            graph_output_value_map = OrderedDict((name_value[0], name_value[1]) for name_value in zip(output_names, pipeline_func_result))\n        else:\n            graph_output_value_map = OrderedDict((output_value.task_output.output_name, output_value) for output_value in pipeline_func_result) # TODO: Fix possible name non-uniqueness (e.g. use task id as prefix or add index to non-unique names)\n    else:\n        raise TypeError('Pipeline must return outputs as tuple or OrderedDict.')\n\n    #Checking the pipeline_func output object types\n    for output_name, output_value in graph_output_value_map.items():\n        if not isinstance(output_value, TaskOutputArgument):\n            raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got \"{output_name}\" = \"{}\".'.format(output_name, str(output_value)))\n\n    if not component_spec.outputs and graph_output_value_map:\n        component_spec.outputs = [OutputSpec(name=output_name, type=output_value.task_output.type) for output_name, output_value in graph_output_value_map.items()]\n\n    component_spec.implementation = GraphImplementation(\n        graph=GraphSpec(\n            tasks=task_map,\n            output_values=graph_output_value_map,\n        )\n    )\n    return component_spec\n", "path": "sdk/python/kfp/components/_python_to_graph_component.py"}], "after_files": [{"content": "# Copyright 2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n__all__ = [\n    'create_graph_component_from_pipeline_func',\n]\n\n\nimport inspect\nfrom collections import OrderedDict\nfrom typing import Callable\n\nfrom . import _components\nfrom ._structures import TaskSpec, ComponentSpec, OutputSpec, GraphInputReference, TaskOutputArgument, GraphImplementation, GraphSpec\nfrom ._naming import _make_name_unique_by_adding_index\nfrom ._python_op import _extract_component_interface\nfrom ._components import _create_task_factory_from_component_spec\n\n\ndef create_graph_component_from_pipeline_func(\n    pipeline_func: Callable,\n    output_component_file: str = None,\n    embed_component_specs: bool = False,\n) -> Callable:\n    '''Experimental! Creates graph component definition from a python pipeline function. The component file can be published for sharing.\n    Pipeline function is a function that only calls component functions and passes outputs to inputs.\n    This feature is experimental and lacks support for some of the DSL features like conditions and loops.\n    Only pipelines consisting of loaded components or python components are currently supported (no manually created ContainerOps or ResourceOps).\n\n    Args:\n        pipeline_func: Python function to convert\n        output_component_file: Path of the file where the component definition will be written. The `component.yaml` file can then be published for sharing.\n        embed_component_specs: Whether to embed component definitions or just reference them. Embedding makes the graph component self-contained. Default is False.\n\n    Returns:\n        A function representing the graph component. The component spec can be accessed using the .component_spec attribute.\n        The function will have the same parameters as the original function.\n        When called, the function will return a task object, corresponding to the graph component.\n        To reference the outputs of the task, use task.outputs[\"Output name\"].\n\n    Example:\n\n        producer_op = load_component_from_file('producer/component.yaml')\n        processor_op = load_component_from_file('processor/component.yaml')\n\n        def pipeline1(pipeline_param_1: int):\n            producer_task = producer_op()\n            processor_task = processor_op(pipeline_param_1, producer_task.outputs['Output 2'])\n\n            return OrderedDict([\n                ('Pipeline output 1', producer_task.outputs['Output 1']),\n                ('Pipeline output 2', processor_task.outputs['Output 2']),\n            ])\n        \n        create_graph_component_from_pipeline_func(pipeline1, output_component_file='pipeline.component.yaml')\n    '''\n    component_spec = create_graph_component_spec_from_pipeline_func(pipeline_func, embed_component_specs)\n    if output_component_file:\n        from pathlib import Path\n        from ._yaml_utils import dump_yaml\n        component_dict = component_spec.to_dict()\n        component_yaml = dump_yaml(component_dict)\n        Path(output_component_file).write_text(component_yaml)\n\n    return _create_task_factory_from_component_spec(component_spec)\n\ndef create_graph_component_spec_from_pipeline_func(pipeline_func: Callable, embed_component_specs: bool = False) -> ComponentSpec:\n\n    component_spec = _extract_component_interface(pipeline_func)\n    # Checking the function parameters - they should not have file passing annotations.\n    input_specs = component_spec.inputs or []\n    for input in input_specs:\n        if input._passing_style:\n            raise TypeError('Graph component function parameter \"{}\" cannot have file-passing annotation \"{}\".'.format(input.name, input._passing_style))\n\n    task_map = OrderedDict() #Preserving task order\n\n    from ._components import _create_task_spec_from_component_and_arguments\n    def task_construction_handler(\n        component_spec,\n        arguments,\n        component_ref,\n    ):\n        task = _create_task_spec_from_component_and_arguments(\n            component_spec=component_spec,\n            arguments=arguments,\n            component_ref=component_ref,\n        )\n\n        #Rewriting task ids so that they're same every time\n        task_id = task.component_ref.spec.name or \"Task\"\n        task_id = _make_name_unique_by_adding_index(task_id, task_map.keys(), ' ')\n        for output_ref in task.outputs.values():\n            output_ref.task_output.task_id = task_id\n            output_ref.task_output.task = None\n        task_map[task_id] = task\n        # Remove the component spec from component reference unless it will make the reference empty or unless explicitly asked by the user\n        if not embed_component_specs and any([task.component_ref.name, task.component_ref.url, task.component_ref.digest]):\n            task.component_ref.spec = None\n\n        return task #The handler is a transformation function, so it must pass the task through.\n\n    # Preparing the pipeline_func arguments\n    # TODO: The key should be original parameter name if different\n    pipeline_func_args = {input.name: GraphInputReference(input_name=input.name).as_argument() for input in input_specs}\n\n    try:\n        #Setting the handler to fix and catch the tasks.\n        # FIX: The handler only hooks container component creation\n        old_handler = _components._container_task_constructor\n        _components._container_task_constructor = task_construction_handler\n        \n        #Calling the pipeline_func with GraphInputArgument instances as arguments \n        pipeline_func_result = pipeline_func(**pipeline_func_args)\n    finally:\n        _components._container_task_constructor = old_handler\n\n\n    # Getting graph outputs\n    output_names = [output.name for output in (component_spec.outputs or [])]\n\n    if len(output_names) == 1 and output_names[0] == 'Output': # TODO: Check whether the NamedTuple syntax was used\n        pipeline_func_result = [pipeline_func_result]\n\n    if isinstance(pipeline_func_result, tuple) and hasattr(pipeline_func_result, '_asdict'): # collections.namedtuple and typing.NamedTuple\n        pipeline_func_result = pipeline_func_result._asdict()\n\n    if isinstance(pipeline_func_result, dict):\n        if output_names:\n            if set(output_names) != set(pipeline_func_result.keys()):\n                raise ValueError('Returned outputs do not match outputs specified in the function signature: {} = {}'.format(str(set(pipeline_func_result.keys())), str(set(output_names))))\n\n    if pipeline_func_result is None:\n        graph_output_value_map = {}\n    elif isinstance(pipeline_func_result, dict):\n        graph_output_value_map = OrderedDict(pipeline_func_result)\n    elif isinstance(pipeline_func_result, (list, tuple)):\n        if output_names:\n            if len(pipeline_func_result) != len(output_names):\n                raise ValueError('Expected {} values from pipeline function, but got {}.'.format(len(output_names), len(pipeline_func_result)))\n            graph_output_value_map = OrderedDict((name_value[0], name_value[1]) for name_value in zip(output_names, pipeline_func_result))\n        else:\n            graph_output_value_map = OrderedDict((output_value.task_output.output_name, output_value) for output_value in pipeline_func_result) # TODO: Fix possible name non-uniqueness (e.g. use task id as prefix or add index to non-unique names)\n    else:\n        raise TypeError('Pipeline must return outputs as tuple or OrderedDict.')\n\n    #Checking the pipeline_func output object types\n    for output_name, output_value in graph_output_value_map.items():\n        if not isinstance(output_value, TaskOutputArgument):\n            raise TypeError('Only TaskOutputArgument instances should be returned from graph component, but got \"{}\" = \"{}\".'.format(output_name, str(output_value)))\n\n    if not component_spec.outputs and graph_output_value_map:\n        component_spec.outputs = [OutputSpec(name=output_name, type=output_value.task_output.type) for output_name, output_value in graph_output_value_map.items()]\n\n    component_spec.implementation = GraphImplementation(\n        graph=GraphSpec(\n            tasks=task_map,\n            output_values=graph_output_value_map,\n        )\n    )\n    return component_spec\n", "path": "sdk/python/kfp/components/_python_to_graph_component.py"}]}