We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[sdk] Getting None as value when float 0 is passed in
When the input for a component is of type float and the value passed in is 0 the component gets it as None. Is this expected? If so what all are the scenarios that this would happen in. Following is a sample pipeline that I used on Google Vertex AI. 

### Environment

*  KFP version: 1.8.2
*  All dependencies version:

kfp                      1.8.2
kfp-pipeline-spec        0.1.11
kfp-server-api           1.4.1


### Steps to reproduce

```
from kfp.v2 import dsl

@dsl.component
def test_func(
    a: float, b: float, c: float = 0.0
) -> NamedTuple("ExampleOutputs", [("sum", float), ("product", float)]):
    """Example function that demonstrates how to return multiple values."""
    print(f"a: {a}")
    print(f"b: {b}")
    print(f"c: {c}")

@dsl.pipeline(
    name="test-kfp-v2",
    description="Test kfp v2 pipeline",
    pipeline_root="",
)
def pipeline(
    a: float = 0.0,
):
   test_func(a, 0.0)
```
Actual

a: None
b: None
c: None

Expected

a: 0
b: 0
c: 0


### Materials and Reference

<!-- Help us debug this issue by providing resources such as: sample code, background context, or links to references. -->

---

<!-- Don't delete message below to encourage users to support your issue! -->
Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


--- 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/v2/components/executor.py`
Content:
```
1 # Copyright 2021 The Kubeflow Authors
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 import inspect
15 import json
16 from typing import Any, Callable, Dict, List, Optional, Union
17 
18 from kfp.v2.components.types import artifact_types, type_annotations
19 
20 
21 class Executor():
22     """Executor executes v2-based Python function components."""
23 
24     def __init__(self, executor_input: Dict, function_to_execute: Callable):
25         self._func = function_to_execute
26         self._input = executor_input
27         self._input_artifacts: Dict[str, artifact_types.Artifact] = {}
28         self._output_artifacts: Dict[str, artifact_types.Artifact] = {}
29 
30         for name, artifacts in self._input.get('inputs',
31                                                {}).get('artifacts', {}).items():
32             artifacts_list = artifacts.get('artifacts')
33             if artifacts_list:
34                 self._input_artifacts[name] = self._make_input_artifact(
35                     artifacts_list[0])
36 
37         for name, artifacts in self._input.get('outputs',
38                                                {}).get('artifacts', {}).items():
39             artifacts_list = artifacts.get('artifacts')
40             if artifacts_list:
41                 self._output_artifacts[name] = self._make_output_artifact(
42                     artifacts_list[0])
43 
44         self._return_annotation = inspect.signature(
45             self._func).return_annotation
46         self._executor_output = {}
47 
48     @classmethod
49     def _make_input_artifact(cls, runtime_artifact: Dict):
50         return artifact_types.create_runtime_artifact(runtime_artifact)
51 
52     @classmethod
53     def _make_output_artifact(cls, runtime_artifact: Dict):
54         import os
55         artifact = artifact_types.create_runtime_artifact(runtime_artifact)
56         os.makedirs(os.path.dirname(artifact.path), exist_ok=True)
57         return artifact
58 
59     def _get_input_artifact(self, name: str):
60         return self._input_artifacts.get(name)
61 
62     def _get_output_artifact(self, name: str):
63         return self._output_artifacts.get(name)
64 
65     def _get_input_parameter_value(self, parameter_name: str,
66                                    parameter_type: Any):
67         parameter = self._input.get('inputs',
68                                     {}).get('parameters',
69                                             {}).get(parameter_name, None)
70         if parameter is None:
71             return None
72 
73         if parameter.get('stringValue') is not None:
74             if parameter_type == str:
75                 return parameter['stringValue']
76             elif parameter_type == bool:
77                 # Use `.lower()` so it can also handle 'True' and 'False' (resulted from
78                 # `str(True)` and `str(False)`, respectively.
79                 return json.loads(parameter['stringValue'].lower())
80             else:
81                 return json.loads(parameter['stringValue'])
82         elif parameter.get('intValue'):
83             return int(parameter['intValue'])
84         elif parameter.get('doubleValue'):
85             return float(parameter['doubleValue'])
86 
87     def _get_output_parameter_path(self, parameter_name: str):
88         parameter = self._input.get('outputs',
89                                     {}).get('parameters',
90                                             {}).get(parameter_name, None)
91         if parameter is None:
92             return None
93 
94         import os
95         path = parameter.get('outputFile', None)
96         if path:
97             os.makedirs(os.path.dirname(path), exist_ok=True)
98         return path
99 
100     def _get_output_artifact_path(self, artifact_name: str):
101         output_artifact = self._output_artifacts.get(artifact_name)
102         if not output_artifact:
103             raise ValueError(
104                 'Failed to get output artifact path for artifact name {}'
105                 .format(artifact_name))
106         return output_artifact.path
107 
108     def _get_input_artifact_path(self, artifact_name: str):
109         input_artifact = self._input_artifacts.get(artifact_name)
110         if not input_artifact:
111             raise ValueError(
112                 'Failed to get input artifact path for artifact name {}'.format(
113                     artifact_name))
114         return input_artifact.path
115 
116     def _write_output_parameter_value(self, name: str,
117                                       value: Union[str, int, float, bool, dict,
118                                                    list, Dict, List]):
119         if type(value) == str:
120             output = {'stringValue': value}
121         elif type(value) == int:
122             output = {'intValue': value}
123         elif type(value) == float:
124             output = {'doubleValue': value}
125         else:
126             # For bool, list, dict, List, Dict, json serialize the value.
127             output = {'stringValue': json.dumps(value)}
128 
129         if not self._executor_output.get('parameters'):
130             self._executor_output['parameters'] = {}
131 
132         self._executor_output['parameters'][name] = output
133 
134     def _write_output_artifact_payload(self, name: str, value: Any):
135         path = self._get_output_artifact_path(name)
136         with open(path, 'w') as f:
137             f.write(str(value))
138 
139     # TODO: extract to a util
140     @classmethod
141     def _get_short_type_name(cls, type_name: str) -> str:
142         """Extracts the short form type name.
143 
144         This method is used for looking up serializer for a given type.
145 
146         For example:
147           typing.List -> List
148           typing.List[int] -> List
149           typing.Dict[str, str] -> Dict
150           List -> List
151           str -> str
152 
153         Args:
154           type_name: The original type name.
155 
156         Returns:
157           The short form type name or the original name if pattern doesn't match.
158         """
159         import re
160         match = re.match('(typing\.)?(?P<type>\w+)(?:\[.+\])?', type_name)
161         if match:
162             return match.group('type')
163         else:
164             return type_name
165 
166     # TODO: merge with type_utils.is_parameter_type
167     @classmethod
168     def _is_parameter(cls, annotation: Any) -> bool:
169         if type(annotation) == type:
170             return annotation in [str, int, float, bool, dict, list]
171 
172         # Annotation could be, for instance `typing.Dict[str, str]`, etc.
173         return cls._get_short_type_name(str(annotation)) in ['Dict', 'List']
174 
175     @classmethod
176     def _is_artifact(cls, annotation: Any) -> bool:
177         if type(annotation) == type:
178             return issubclass(annotation, artifact_types.Artifact)
179         return False
180 
181     @classmethod
182     def _is_named_tuple(cls, annotation: Any) -> bool:
183         if type(annotation) == type:
184             return issubclass(annotation, tuple) and hasattr(
185                 annotation, '_fields') and hasattr(annotation,
186                                                    '__annotations__')
187         return False
188 
189     def _handle_single_return_value(self, output_name: str,
190                                     annotation_type: Any, return_value: Any):
191         if self._is_parameter(annotation_type):
192             origin_type = getattr(annotation_type, '__origin__', None) or annotation_type
193             if not isinstance(return_value, origin_type):
194                 raise ValueError(
195                     'Function `{}` returned value of type {}; want type {}'
196                     .format(self._func.__name__, type(return_value),
197                             origin_type))
198             self._write_output_parameter_value(output_name, return_value)
199         elif self._is_artifact(annotation_type):
200             self._write_output_artifact_payload(output_name, return_value)
201         else:
202             raise RuntimeError(
203                 'Unknown return type: {}. Must be one of `str`, `int`, `float`, or a'
204                 ' subclass of `Artifact`'.format(annotation_type))
205 
206     def _write_executor_output(self, func_output: Optional[Any] = None):
207         if self._output_artifacts:
208             self._executor_output['artifacts'] = {}
209 
210         for name, artifact in self._output_artifacts.items():
211             runtime_artifact = {
212                 'name': artifact.name,
213                 'uri': artifact.uri,
214                 'metadata': artifact.metadata,
215             }
216             artifacts_list = {'artifacts': [runtime_artifact]}
217 
218             self._executor_output['artifacts'][name] = artifacts_list
219 
220         if func_output is not None:
221             if self._is_parameter(self._return_annotation) or self._is_artifact(
222                     self._return_annotation):
223                 # Note: single output is named `Output` in component.yaml.
224                 self._handle_single_return_value('Output',
225                                                  self._return_annotation,
226                                                  func_output)
227             elif self._is_named_tuple(self._return_annotation):
228                 if len(self._return_annotation._fields) != len(func_output):
229                     raise RuntimeError(
230                         'Expected {} return values from function `{}`, got {}'
231                         .format(
232                             len(self._return_annotation._fields),
233                             self._func.__name__, len(func_output)))
234                 for i in range(len(self._return_annotation._fields)):
235                     field = self._return_annotation._fields[i]
236                     field_type = self._return_annotation.__annotations__[field]
237                     if type(func_output) == tuple:
238                         field_value = func_output[i]
239                     else:
240                         field_value = getattr(func_output, field)
241                     self._handle_single_return_value(field, field_type,
242                                                      field_value)
243             else:
244                 raise RuntimeError(
245                     'Unknown return type: {}. Must be one of `str`, `int`, `float`, a'
246                     ' subclass of `Artifact`, or a NamedTuple collection of these types.'
247                     .format(self._return_annotation))
248 
249         import os
250         os.makedirs(
251             os.path.dirname(self._input['outputs']['outputFile']),
252             exist_ok=True)
253         with open(self._input['outputs']['outputFile'], 'w') as f:
254             f.write(json.dumps(self._executor_output))
255 
256     def execute(self):
257         annotations = inspect.getfullargspec(self._func).annotations
258 
259         # Function arguments.
260         func_kwargs = {}
261 
262         for k, v in annotations.items():
263             if k == 'return':
264                 continue
265 
266             # Annotations for parameter types could be written as, for example,
267             # `Optional[str]`. In this case, we need to strip off the part
268             # `Optional[]` to get the actual parameter type.
269             v = type_annotations.maybe_strip_optional_from_annotation(v)
270 
271             if self._is_parameter(v):
272                 func_kwargs[k] = self._get_input_parameter_value(k, v)
273 
274             if type_annotations.is_artifact_annotation(v):
275                 if type_annotations.is_input_artifact(v):
276                     func_kwargs[k] = self._get_input_artifact(k)
277                 if type_annotations.is_output_artifact(v):
278                     func_kwargs[k] = self._get_output_artifact(k)
279 
280             elif isinstance(v, type_annotations.OutputPath):
281                 if self._is_parameter(v.type):
282                     func_kwargs[k] = self._get_output_parameter_path(k)
283                 else:
284                     func_kwargs[k] = self._get_output_artifact_path(k)
285             elif isinstance(v, type_annotations.InputPath):
286                 func_kwargs[k] = self._get_input_artifact_path(k)
287 
288         result = self._func(**func_kwargs)
289         self._write_executor_output(result)
290
```
--- 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/v2/components/executor.py b/sdk/python/kfp/v2/components/executor.py
--- a/sdk/python/kfp/v2/components/executor.py
+++ b/sdk/python/kfp/v2/components/executor.py
@@ -79,9 +79,9 @@
                 return json.loads(parameter['stringValue'].lower())
             else:
                 return json.loads(parameter['stringValue'])
-        elif parameter.get('intValue'):
+        elif parameter.get('intValue') is not None:
             return int(parameter['intValue'])
-        elif parameter.get('doubleValue'):
+        elif parameter.get('doubleValue') is not None:
             return float(parameter['doubleValue'])
 
     def _get_output_parameter_path(self, parameter_name: str):
@@ -189,7 +189,8 @@
     def _handle_single_return_value(self, output_name: str,
                                     annotation_type: Any, return_value: Any):
         if self._is_parameter(annotation_type):
-            origin_type = getattr(annotation_type, '__origin__', None) or annotation_type
+            origin_type = getattr(annotation_type, '__origin__',
+                                  None) or annotation_type
             if not isinstance(return_value, origin_type):
                 raise ValueError(
                     'Function `{}` returned value of type {}; want type {}'

{"golden_diff": "diff --git a/sdk/python/kfp/v2/components/executor.py b/sdk/python/kfp/v2/components/executor.py\n--- a/sdk/python/kfp/v2/components/executor.py\n+++ b/sdk/python/kfp/v2/components/executor.py\n@@ -79,9 +79,9 @@\n                 return json.loads(parameter['stringValue'].lower())\n             else:\n                 return json.loads(parameter['stringValue'])\n-        elif parameter.get('intValue'):\n+        elif parameter.get('intValue') is not None:\n             return int(parameter['intValue'])\n-        elif parameter.get('doubleValue'):\n+        elif parameter.get('doubleValue') is not None:\n             return float(parameter['doubleValue'])\n \n     def _get_output_parameter_path(self, parameter_name: str):\n@@ -189,7 +189,8 @@\n     def _handle_single_return_value(self, output_name: str,\n                                     annotation_type: Any, return_value: Any):\n         if self._is_parameter(annotation_type):\n-            origin_type = getattr(annotation_type, '__origin__', None) or annotation_type\n+            origin_type = getattr(annotation_type, '__origin__',\n+                                  None) or annotation_type\n             if not isinstance(return_value, origin_type):\n                 raise ValueError(\n                     'Function `{}` returned value of type {}; want type {}'\n", "issue": "[sdk] Getting None as value when float 0 is passed in\nWhen the input for a component is of type float and the value passed in is 0 the component gets it as None. Is this expected? If so what all are the scenarios that this would happen in. Following is a sample pipeline that I used on Google Vertex AI. \r\n\r\n### Environment\r\n\r\n*  KFP version: 1.8.2\r\n*  All dependencies version:\r\n\r\nkfp                      1.8.2\r\nkfp-pipeline-spec        0.1.11\r\nkfp-server-api           1.4.1\r\n\r\n\r\n### Steps to reproduce\r\n\r\n```\r\nfrom kfp.v2 import dsl\r\n\r\n@dsl.component\r\ndef test_func(\r\n    a: float, b: float, c: float = 0.0\r\n) -> NamedTuple(\"ExampleOutputs\", [(\"sum\", float), (\"product\", float)]):\r\n    \"\"\"Example function that demonstrates how to return multiple values.\"\"\"\r\n    print(f\"a: {a}\")\r\n    print(f\"b: {b}\")\r\n    print(f\"c: {c}\")\r\n\r\n@dsl.pipeline(\r\n    name=\"test-kfp-v2\",\r\n    description=\"Test kfp v2 pipeline\",\r\n    pipeline_root=\"\",\r\n)\r\ndef pipeline(\r\n    a: float = 0.0,\r\n):\r\n   test_func(a, 0.0)\r\n```\r\nActual\r\n\r\na: None\r\nb: None\r\nc: None\r\n\r\nExpected\r\n\r\na: 0\r\nb: 0\r\nc: 0\r\n\r\n\r\n### Materials and Reference\r\n\r\n<!-- Help us debug this issue by providing resources such as: sample code, background context, or links to references. -->\r\n\r\n---\r\n\r\n<!-- Don't delete message below to encourage users to support your issue! -->\r\nImpacted by this bug? Give it a \ud83d\udc4d. We prioritise the issues with the most \ud83d\udc4d.\r\n\n", "before_files": [{"content": "# Copyright 2021 The Kubeflow Authors\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.\nimport inspect\nimport json\nfrom typing import Any, Callable, Dict, List, Optional, Union\n\nfrom kfp.v2.components.types import artifact_types, type_annotations\n\n\nclass Executor():\n    \"\"\"Executor executes v2-based Python function components.\"\"\"\n\n    def __init__(self, executor_input: Dict, function_to_execute: Callable):\n        self._func = function_to_execute\n        self._input = executor_input\n        self._input_artifacts: Dict[str, artifact_types.Artifact] = {}\n        self._output_artifacts: Dict[str, artifact_types.Artifact] = {}\n\n        for name, artifacts in self._input.get('inputs',\n                                               {}).get('artifacts', {}).items():\n            artifacts_list = artifacts.get('artifacts')\n            if artifacts_list:\n                self._input_artifacts[name] = self._make_input_artifact(\n                    artifacts_list[0])\n\n        for name, artifacts in self._input.get('outputs',\n                                               {}).get('artifacts', {}).items():\n            artifacts_list = artifacts.get('artifacts')\n            if artifacts_list:\n                self._output_artifacts[name] = self._make_output_artifact(\n                    artifacts_list[0])\n\n        self._return_annotation = inspect.signature(\n            self._func).return_annotation\n        self._executor_output = {}\n\n    @classmethod\n    def _make_input_artifact(cls, runtime_artifact: Dict):\n        return artifact_types.create_runtime_artifact(runtime_artifact)\n\n    @classmethod\n    def _make_output_artifact(cls, runtime_artifact: Dict):\n        import os\n        artifact = artifact_types.create_runtime_artifact(runtime_artifact)\n        os.makedirs(os.path.dirname(artifact.path), exist_ok=True)\n        return artifact\n\n    def _get_input_artifact(self, name: str):\n        return self._input_artifacts.get(name)\n\n    def _get_output_artifact(self, name: str):\n        return self._output_artifacts.get(name)\n\n    def _get_input_parameter_value(self, parameter_name: str,\n                                   parameter_type: Any):\n        parameter = self._input.get('inputs',\n                                    {}).get('parameters',\n                                            {}).get(parameter_name, None)\n        if parameter is None:\n            return None\n\n        if parameter.get('stringValue') is not None:\n            if parameter_type == str:\n                return parameter['stringValue']\n            elif parameter_type == bool:\n                # Use `.lower()` so it can also handle 'True' and 'False' (resulted from\n                # `str(True)` and `str(False)`, respectively.\n                return json.loads(parameter['stringValue'].lower())\n            else:\n                return json.loads(parameter['stringValue'])\n        elif parameter.get('intValue'):\n            return int(parameter['intValue'])\n        elif parameter.get('doubleValue'):\n            return float(parameter['doubleValue'])\n\n    def _get_output_parameter_path(self, parameter_name: str):\n        parameter = self._input.get('outputs',\n                                    {}).get('parameters',\n                                            {}).get(parameter_name, None)\n        if parameter is None:\n            return None\n\n        import os\n        path = parameter.get('outputFile', None)\n        if path:\n            os.makedirs(os.path.dirname(path), exist_ok=True)\n        return path\n\n    def _get_output_artifact_path(self, artifact_name: str):\n        output_artifact = self._output_artifacts.get(artifact_name)\n        if not output_artifact:\n            raise ValueError(\n                'Failed to get output artifact path for artifact name {}'\n                .format(artifact_name))\n        return output_artifact.path\n\n    def _get_input_artifact_path(self, artifact_name: str):\n        input_artifact = self._input_artifacts.get(artifact_name)\n        if not input_artifact:\n            raise ValueError(\n                'Failed to get input artifact path for artifact name {}'.format(\n                    artifact_name))\n        return input_artifact.path\n\n    def _write_output_parameter_value(self, name: str,\n                                      value: Union[str, int, float, bool, dict,\n                                                   list, Dict, List]):\n        if type(value) == str:\n            output = {'stringValue': value}\n        elif type(value) == int:\n            output = {'intValue': value}\n        elif type(value) == float:\n            output = {'doubleValue': value}\n        else:\n            # For bool, list, dict, List, Dict, json serialize the value.\n            output = {'stringValue': json.dumps(value)}\n\n        if not self._executor_output.get('parameters'):\n            self._executor_output['parameters'] = {}\n\n        self._executor_output['parameters'][name] = output\n\n    def _write_output_artifact_payload(self, name: str, value: Any):\n        path = self._get_output_artifact_path(name)\n        with open(path, 'w') as f:\n            f.write(str(value))\n\n    # TODO: extract to a util\n    @classmethod\n    def _get_short_type_name(cls, type_name: str) -> str:\n        \"\"\"Extracts the short form type name.\n\n        This method is used for looking up serializer for a given type.\n\n        For example:\n          typing.List -> List\n          typing.List[int] -> List\n          typing.Dict[str, str] -> Dict\n          List -> List\n          str -> str\n\n        Args:\n          type_name: The original type name.\n\n        Returns:\n          The short form type name or the original name if pattern doesn't match.\n        \"\"\"\n        import re\n        match = re.match('(typing\\.)?(?P<type>\\w+)(?:\\[.+\\])?', type_name)\n        if match:\n            return match.group('type')\n        else:\n            return type_name\n\n    # TODO: merge with type_utils.is_parameter_type\n    @classmethod\n    def _is_parameter(cls, annotation: Any) -> bool:\n        if type(annotation) == type:\n            return annotation in [str, int, float, bool, dict, list]\n\n        # Annotation could be, for instance `typing.Dict[str, str]`, etc.\n        return cls._get_short_type_name(str(annotation)) in ['Dict', 'List']\n\n    @classmethod\n    def _is_artifact(cls, annotation: Any) -> bool:\n        if type(annotation) == type:\n            return issubclass(annotation, artifact_types.Artifact)\n        return False\n\n    @classmethod\n    def _is_named_tuple(cls, annotation: Any) -> bool:\n        if type(annotation) == type:\n            return issubclass(annotation, tuple) and hasattr(\n                annotation, '_fields') and hasattr(annotation,\n                                                   '__annotations__')\n        return False\n\n    def _handle_single_return_value(self, output_name: str,\n                                    annotation_type: Any, return_value: Any):\n        if self._is_parameter(annotation_type):\n            origin_type = getattr(annotation_type, '__origin__', None) or annotation_type\n            if not isinstance(return_value, origin_type):\n                raise ValueError(\n                    'Function `{}` returned value of type {}; want type {}'\n                    .format(self._func.__name__, type(return_value),\n                            origin_type))\n            self._write_output_parameter_value(output_name, return_value)\n        elif self._is_artifact(annotation_type):\n            self._write_output_artifact_payload(output_name, return_value)\n        else:\n            raise RuntimeError(\n                'Unknown return type: {}. Must be one of `str`, `int`, `float`, or a'\n                ' subclass of `Artifact`'.format(annotation_type))\n\n    def _write_executor_output(self, func_output: Optional[Any] = None):\n        if self._output_artifacts:\n            self._executor_output['artifacts'] = {}\n\n        for name, artifact in self._output_artifacts.items():\n            runtime_artifact = {\n                'name': artifact.name,\n                'uri': artifact.uri,\n                'metadata': artifact.metadata,\n            }\n            artifacts_list = {'artifacts': [runtime_artifact]}\n\n            self._executor_output['artifacts'][name] = artifacts_list\n\n        if func_output is not None:\n            if self._is_parameter(self._return_annotation) or self._is_artifact(\n                    self._return_annotation):\n                # Note: single output is named `Output` in component.yaml.\n                self._handle_single_return_value('Output',\n                                                 self._return_annotation,\n                                                 func_output)\n            elif self._is_named_tuple(self._return_annotation):\n                if len(self._return_annotation._fields) != len(func_output):\n                    raise RuntimeError(\n                        'Expected {} return values from function `{}`, got {}'\n                        .format(\n                            len(self._return_annotation._fields),\n                            self._func.__name__, len(func_output)))\n                for i in range(len(self._return_annotation._fields)):\n                    field = self._return_annotation._fields[i]\n                    field_type = self._return_annotation.__annotations__[field]\n                    if type(func_output) == tuple:\n                        field_value = func_output[i]\n                    else:\n                        field_value = getattr(func_output, field)\n                    self._handle_single_return_value(field, field_type,\n                                                     field_value)\n            else:\n                raise RuntimeError(\n                    'Unknown return type: {}. Must be one of `str`, `int`, `float`, a'\n                    ' subclass of `Artifact`, or a NamedTuple collection of these types.'\n                    .format(self._return_annotation))\n\n        import os\n        os.makedirs(\n            os.path.dirname(self._input['outputs']['outputFile']),\n            exist_ok=True)\n        with open(self._input['outputs']['outputFile'], 'w') as f:\n            f.write(json.dumps(self._executor_output))\n\n    def execute(self):\n        annotations = inspect.getfullargspec(self._func).annotations\n\n        # Function arguments.\n        func_kwargs = {}\n\n        for k, v in annotations.items():\n            if k == 'return':\n                continue\n\n            # Annotations for parameter types could be written as, for example,\n            # `Optional[str]`. In this case, we need to strip off the part\n            # `Optional[]` to get the actual parameter type.\n            v = type_annotations.maybe_strip_optional_from_annotation(v)\n\n            if self._is_parameter(v):\n                func_kwargs[k] = self._get_input_parameter_value(k, v)\n\n            if type_annotations.is_artifact_annotation(v):\n                if type_annotations.is_input_artifact(v):\n                    func_kwargs[k] = self._get_input_artifact(k)\n                if type_annotations.is_output_artifact(v):\n                    func_kwargs[k] = self._get_output_artifact(k)\n\n            elif isinstance(v, type_annotations.OutputPath):\n                if self._is_parameter(v.type):\n                    func_kwargs[k] = self._get_output_parameter_path(k)\n                else:\n                    func_kwargs[k] = self._get_output_artifact_path(k)\n            elif isinstance(v, type_annotations.InputPath):\n                func_kwargs[k] = self._get_input_artifact_path(k)\n\n        result = self._func(**func_kwargs)\n        self._write_executor_output(result)\n", "path": "sdk/python/kfp/v2/components/executor.py"}], "after_files": [{"content": "# Copyright 2021 The Kubeflow Authors\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.\nimport inspect\nimport json\nfrom typing import Any, Callable, Dict, List, Optional, Union\n\nfrom kfp.v2.components.types import artifact_types, type_annotations\n\n\nclass Executor():\n    \"\"\"Executor executes v2-based Python function components.\"\"\"\n\n    def __init__(self, executor_input: Dict, function_to_execute: Callable):\n        self._func = function_to_execute\n        self._input = executor_input\n        self._input_artifacts: Dict[str, artifact_types.Artifact] = {}\n        self._output_artifacts: Dict[str, artifact_types.Artifact] = {}\n\n        for name, artifacts in self._input.get('inputs',\n                                               {}).get('artifacts', {}).items():\n            artifacts_list = artifacts.get('artifacts')\n            if artifacts_list:\n                self._input_artifacts[name] = self._make_input_artifact(\n                    artifacts_list[0])\n\n        for name, artifacts in self._input.get('outputs',\n                                               {}).get('artifacts', {}).items():\n            artifacts_list = artifacts.get('artifacts')\n            if artifacts_list:\n                self._output_artifacts[name] = self._make_output_artifact(\n                    artifacts_list[0])\n\n        self._return_annotation = inspect.signature(\n            self._func).return_annotation\n        self._executor_output = {}\n\n    @classmethod\n    def _make_input_artifact(cls, runtime_artifact: Dict):\n        return artifact_types.create_runtime_artifact(runtime_artifact)\n\n    @classmethod\n    def _make_output_artifact(cls, runtime_artifact: Dict):\n        import os\n        artifact = artifact_types.create_runtime_artifact(runtime_artifact)\n        os.makedirs(os.path.dirname(artifact.path), exist_ok=True)\n        return artifact\n\n    def _get_input_artifact(self, name: str):\n        return self._input_artifacts.get(name)\n\n    def _get_output_artifact(self, name: str):\n        return self._output_artifacts.get(name)\n\n    def _get_input_parameter_value(self, parameter_name: str,\n                                   parameter_type: Any):\n        parameter = self._input.get('inputs',\n                                    {}).get('parameters',\n                                            {}).get(parameter_name, None)\n        if parameter is None:\n            return None\n\n        if parameter.get('stringValue') is not None:\n            if parameter_type == str:\n                return parameter['stringValue']\n            elif parameter_type == bool:\n                # Use `.lower()` so it can also handle 'True' and 'False' (resulted from\n                # `str(True)` and `str(False)`, respectively.\n                return json.loads(parameter['stringValue'].lower())\n            else:\n                return json.loads(parameter['stringValue'])\n        elif parameter.get('intValue') is not None:\n            return int(parameter['intValue'])\n        elif parameter.get('doubleValue') is not None:\n            return float(parameter['doubleValue'])\n\n    def _get_output_parameter_path(self, parameter_name: str):\n        parameter = self._input.get('outputs',\n                                    {}).get('parameters',\n                                            {}).get(parameter_name, None)\n        if parameter is None:\n            return None\n\n        import os\n        path = parameter.get('outputFile', None)\n        if path:\n            os.makedirs(os.path.dirname(path), exist_ok=True)\n        return path\n\n    def _get_output_artifact_path(self, artifact_name: str):\n        output_artifact = self._output_artifacts.get(artifact_name)\n        if not output_artifact:\n            raise ValueError(\n                'Failed to get output artifact path for artifact name {}'\n                .format(artifact_name))\n        return output_artifact.path\n\n    def _get_input_artifact_path(self, artifact_name: str):\n        input_artifact = self._input_artifacts.get(artifact_name)\n        if not input_artifact:\n            raise ValueError(\n                'Failed to get input artifact path for artifact name {}'.format(\n                    artifact_name))\n        return input_artifact.path\n\n    def _write_output_parameter_value(self, name: str,\n                                      value: Union[str, int, float, bool, dict,\n                                                   list, Dict, List]):\n        if type(value) == str:\n            output = {'stringValue': value}\n        elif type(value) == int:\n            output = {'intValue': value}\n        elif type(value) == float:\n            output = {'doubleValue': value}\n        else:\n            # For bool, list, dict, List, Dict, json serialize the value.\n            output = {'stringValue': json.dumps(value)}\n\n        if not self._executor_output.get('parameters'):\n            self._executor_output['parameters'] = {}\n\n        self._executor_output['parameters'][name] = output\n\n    def _write_output_artifact_payload(self, name: str, value: Any):\n        path = self._get_output_artifact_path(name)\n        with open(path, 'w') as f:\n            f.write(str(value))\n\n    # TODO: extract to a util\n    @classmethod\n    def _get_short_type_name(cls, type_name: str) -> str:\n        \"\"\"Extracts the short form type name.\n\n        This method is used for looking up serializer for a given type.\n\n        For example:\n          typing.List -> List\n          typing.List[int] -> List\n          typing.Dict[str, str] -> Dict\n          List -> List\n          str -> str\n\n        Args:\n          type_name: The original type name.\n\n        Returns:\n          The short form type name or the original name if pattern doesn't match.\n        \"\"\"\n        import re\n        match = re.match('(typing\\.)?(?P<type>\\w+)(?:\\[.+\\])?', type_name)\n        if match:\n            return match.group('type')\n        else:\n            return type_name\n\n    # TODO: merge with type_utils.is_parameter_type\n    @classmethod\n    def _is_parameter(cls, annotation: Any) -> bool:\n        if type(annotation) == type:\n            return annotation in [str, int, float, bool, dict, list]\n\n        # Annotation could be, for instance `typing.Dict[str, str]`, etc.\n        return cls._get_short_type_name(str(annotation)) in ['Dict', 'List']\n\n    @classmethod\n    def _is_artifact(cls, annotation: Any) -> bool:\n        if type(annotation) == type:\n            return issubclass(annotation, artifact_types.Artifact)\n        return False\n\n    @classmethod\n    def _is_named_tuple(cls, annotation: Any) -> bool:\n        if type(annotation) == type:\n            return issubclass(annotation, tuple) and hasattr(\n                annotation, '_fields') and hasattr(annotation,\n                                                   '__annotations__')\n        return False\n\n    def _handle_single_return_value(self, output_name: str,\n                                    annotation_type: Any, return_value: Any):\n        if self._is_parameter(annotation_type):\n            origin_type = getattr(annotation_type, '__origin__',\n                                  None) or annotation_type\n            if not isinstance(return_value, origin_type):\n                raise ValueError(\n                    'Function `{}` returned value of type {}; want type {}'\n                    .format(self._func.__name__, type(return_value),\n                            origin_type))\n            self._write_output_parameter_value(output_name, return_value)\n        elif self._is_artifact(annotation_type):\n            self._write_output_artifact_payload(output_name, return_value)\n        else:\n            raise RuntimeError(\n                'Unknown return type: {}. Must be one of `str`, `int`, `float`, or a'\n                ' subclass of `Artifact`'.format(annotation_type))\n\n    def _write_executor_output(self, func_output: Optional[Any] = None):\n        if self._output_artifacts:\n            self._executor_output['artifacts'] = {}\n\n        for name, artifact in self._output_artifacts.items():\n            runtime_artifact = {\n                'name': artifact.name,\n                'uri': artifact.uri,\n                'metadata': artifact.metadata,\n            }\n            artifacts_list = {'artifacts': [runtime_artifact]}\n\n            self._executor_output['artifacts'][name] = artifacts_list\n\n        if func_output is not None:\n            if self._is_parameter(self._return_annotation) or self._is_artifact(\n                    self._return_annotation):\n                # Note: single output is named `Output` in component.yaml.\n                self._handle_single_return_value('Output',\n                                                 self._return_annotation,\n                                                 func_output)\n            elif self._is_named_tuple(self._return_annotation):\n                if len(self._return_annotation._fields) != len(func_output):\n                    raise RuntimeError(\n                        'Expected {} return values from function `{}`, got {}'\n                        .format(\n                            len(self._return_annotation._fields),\n                            self._func.__name__, len(func_output)))\n                for i in range(len(self._return_annotation._fields)):\n                    field = self._return_annotation._fields[i]\n                    field_type = self._return_annotation.__annotations__[field]\n                    if type(func_output) == tuple:\n                        field_value = func_output[i]\n                    else:\n                        field_value = getattr(func_output, field)\n                    self._handle_single_return_value(field, field_type,\n                                                     field_value)\n            else:\n                raise RuntimeError(\n                    'Unknown return type: {}. Must be one of `str`, `int`, `float`, a'\n                    ' subclass of `Artifact`, or a NamedTuple collection of these types.'\n                    .format(self._return_annotation))\n\n        import os\n        os.makedirs(\n            os.path.dirname(self._input['outputs']['outputFile']),\n            exist_ok=True)\n        with open(self._input['outputs']['outputFile'], 'w') as f:\n            f.write(json.dumps(self._executor_output))\n\n    def execute(self):\n        annotations = inspect.getfullargspec(self._func).annotations\n\n        # Function arguments.\n        func_kwargs = {}\n\n        for k, v in annotations.items():\n            if k == 'return':\n                continue\n\n            # Annotations for parameter types could be written as, for example,\n            # `Optional[str]`. In this case, we need to strip off the part\n            # `Optional[]` to get the actual parameter type.\n            v = type_annotations.maybe_strip_optional_from_annotation(v)\n\n            if self._is_parameter(v):\n                func_kwargs[k] = self._get_input_parameter_value(k, v)\n\n            if type_annotations.is_artifact_annotation(v):\n                if type_annotations.is_input_artifact(v):\n                    func_kwargs[k] = self._get_input_artifact(k)\n                if type_annotations.is_output_artifact(v):\n                    func_kwargs[k] = self._get_output_artifact(k)\n\n            elif isinstance(v, type_annotations.OutputPath):\n                if self._is_parameter(v.type):\n                    func_kwargs[k] = self._get_output_parameter_path(k)\n                else:\n                    func_kwargs[k] = self._get_output_artifact_path(k)\n            elif isinstance(v, type_annotations.InputPath):\n                func_kwargs[k] = self._get_input_artifact_path(k)\n\n        result = self._func(**func_kwargs)\n        self._write_executor_output(result)\n", "path": "sdk/python/kfp/v2/components/executor.py"}]}