We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
SDK - Support placeholders in task display names
Fixes https://github.com/kubeflow/pipelines/issues/4145

--- 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/compiler/_op_to_template.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 import json
16 import re
17 import warnings
18 import yaml
19 from collections import OrderedDict
20 from typing import Union, List, Any, Callable, TypeVar, Dict
21 
22 from ._k8s_helper import convert_k8s_obj_to_json
23 from .. import dsl
24 from ..dsl._container_op import BaseOp
25 
26 
27 # generics
28 T = TypeVar('T')
29 
30 
31 def _process_obj(obj: Any, map_to_tmpl_var: dict):
32     """Recursively sanitize and replace any PipelineParam (instances and serialized strings)
33     in the object with the corresponding template variables
34     (i.e. '{{inputs.parameters.<PipelineParam.full_name>}}').
35     
36     Args:
37       obj: any obj that may have PipelineParam
38       map_to_tmpl_var: a dict that maps an unsanitized pipeline
39                        params signature into a template var
40     """
41     # serialized str might be unsanitized
42     if isinstance(obj, str):
43         # get signature
44         param_tuples = dsl.match_serialized_pipelineparam(obj)
45         if not param_tuples:
46             return obj
47         # replace all unsanitized signature with template var
48         for param_tuple in param_tuples:
49             obj = re.sub(param_tuple.pattern, map_to_tmpl_var[param_tuple.pattern], obj)
50 
51     # list
52     if isinstance(obj, list):
53         return [_process_obj(item, map_to_tmpl_var) for item in obj]
54 
55     # tuple
56     if isinstance(obj, tuple):
57         return tuple((_process_obj(item, map_to_tmpl_var) for item in obj))
58 
59     # dict
60     if isinstance(obj, dict):
61         return {
62             _process_obj(key, map_to_tmpl_var): _process_obj(value, map_to_tmpl_var)
63             for key, value in obj.items()
64         }
65 
66     # pipelineparam
67     if isinstance(obj, dsl.PipelineParam):
68         # if not found in unsanitized map, then likely to be sanitized
69         return map_to_tmpl_var.get(str(obj), '{{inputs.parameters.%s}}' % obj.full_name)
70 
71     # k8s objects (generated from swaggercodegen)
72     if hasattr(obj, 'attribute_map') and isinstance(obj.attribute_map, dict):
73         # process everything inside recursively
74         for key in obj.attribute_map.keys():
75             setattr(obj, key, _process_obj(getattr(obj, key), map_to_tmpl_var))
76         # return json representation of the k8s obj
77         return convert_k8s_obj_to_json(obj)
78 
79     # do nothing
80     return obj
81 
82 
83 def _process_base_ops(op: BaseOp):
84     """Recursively go through the attrs listed in `attrs_with_pipelineparams`
85     and sanitize and replace pipeline params with template var string.
86 
87     Returns a processed `BaseOp`.
88 
89     NOTE this is an in-place update to `BaseOp`'s attributes (i.e. the ones
90     specified in `attrs_with_pipelineparams`, all `PipelineParam` are replaced
91     with the corresponding template variable strings).
92 
93     Args:
94         op {BaseOp}: class that inherits from BaseOp
95 
96     Returns:
97         BaseOp
98     """
99 
100     # map param's (unsanitized pattern or serialized str pattern) -> input param var str
101     map_to_tmpl_var = {
102         (param.pattern or str(param)): '{{inputs.parameters.%s}}' % param.full_name
103         for param in op.inputs
104     }
105 
106     # process all attr with pipelineParams except inputs and outputs parameters
107     for key in op.attrs_with_pipelineparams:
108         setattr(op, key, _process_obj(getattr(op, key), map_to_tmpl_var))
109 
110     return op
111 
112 
113 def _parameters_to_json(params: List[dsl.PipelineParam]):
114     """Converts a list of PipelineParam into an argo `parameter` JSON obj."""
115     _to_json = (lambda param: dict(name=param.full_name, value=param.value)
116                 if param.value else dict(name=param.full_name))
117     params = [_to_json(param) for param in params]
118     # Sort to make the results deterministic.
119     params.sort(key=lambda x: x['name'])
120     return params
121 
122 
123 def _inputs_to_json(
124     inputs_params: List[dsl.PipelineParam],
125     input_artifact_paths: Dict[str, str] = None,
126     artifact_arguments: Dict[str, str] = None,
127 ) -> Dict[str, Dict]:
128     """Converts a list of PipelineParam into an argo `inputs` JSON obj."""
129     parameters = _parameters_to_json(inputs_params)
130 
131     # Building the input artifacts section
132     artifacts = []
133     for name, path in (input_artifact_paths or {}).items():
134         artifact = {'name': name, 'path': path}
135         if name in artifact_arguments: # The arguments should be compiled as DAG task arguments, not template's default values, but in the current DSL-compiler implementation it's too hard to make that work when passing artifact references.
136             artifact['raw'] = {'data': str(artifact_arguments[name])}
137         artifacts.append(artifact)
138     artifacts.sort(key=lambda x: x['name']) #Stabilizing the input artifact ordering
139 
140     inputs_dict = {}
141     if parameters:
142         inputs_dict['parameters'] = parameters
143     if artifacts:
144         inputs_dict['artifacts'] = artifacts
145     return inputs_dict
146 
147 
148 def _outputs_to_json(op: BaseOp,
149                      outputs: Dict[str, dsl.PipelineParam],
150                      param_outputs: Dict[str, str],
151                      output_artifacts: List[dict]):
152     """Creates an argo `outputs` JSON obj."""
153     if isinstance(op, dsl.ResourceOp):
154         value_from_key = "jsonPath"
155     else:
156         value_from_key = "path"
157     output_parameters = []
158     for param in set(outputs.values()):  # set() dedupes output references
159         output_parameters.append({
160             'name': param.full_name,
161             'valueFrom': {
162                 value_from_key: param_outputs[param.name]
163             }
164         })
165     output_parameters.sort(key=lambda x: x['name'])
166     ret = {}
167     if output_parameters:
168         ret['parameters'] = output_parameters
169     if output_artifacts:
170         ret['artifacts'] = output_artifacts
171 
172     return ret
173 
174 
175 # TODO: generate argo python classes from swagger and use convert_k8s_obj_to_json??
176 def _op_to_template(op: BaseOp):
177     """Generate template given an operator inherited from BaseOp."""
178 
179     # NOTE in-place update to BaseOp
180     # replace all PipelineParams with template var strings
181     processed_op = _process_base_ops(op)
182 
183     if isinstance(op, dsl.ContainerOp):
184         # default output artifacts
185         output_artifact_paths = OrderedDict(op.output_artifact_paths)
186         # This should have been as easy as output_artifact_paths.update(op.file_outputs), but the _outputs_to_json function changes the output names and we must do the same here, so that the names are the same
187         output_artifact_paths.update(sorted(((param.full_name, processed_op.file_outputs[param.name]) for param in processed_op.outputs.values()), key=lambda x: x[0]))
188 
189         output_artifacts = [
190             {'name': name, 'path': path}
191             for name, path in output_artifact_paths.items()
192         ]
193 
194         # workflow template
195         template = {
196             'name': processed_op.name,
197             'container': convert_k8s_obj_to_json(
198                 processed_op.container
199             )
200         }
201     elif isinstance(op, dsl.ResourceOp):
202         # no output artifacts
203         output_artifacts = []
204 
205         # workflow template
206         processed_op.resource["manifest"] = yaml.dump(
207             convert_k8s_obj_to_json(processed_op.k8s_resource),
208             default_flow_style=False
209         )
210         template = {
211             'name': processed_op.name,
212             'resource': convert_k8s_obj_to_json(
213                 processed_op.resource
214             )
215         }
216 
217     # inputs
218     input_artifact_paths = processed_op.input_artifact_paths if isinstance(processed_op, dsl.ContainerOp) else None
219     artifact_arguments = processed_op.artifact_arguments if isinstance(processed_op, dsl.ContainerOp) else None
220     inputs = _inputs_to_json(processed_op.inputs, input_artifact_paths, artifact_arguments)
221     if inputs:
222         template['inputs'] = inputs
223 
224     # outputs
225     if isinstance(op, dsl.ContainerOp):
226         param_outputs = processed_op.file_outputs
227     elif isinstance(op, dsl.ResourceOp):
228         param_outputs = processed_op.attribute_outputs
229     outputs_dict = _outputs_to_json(op, processed_op.outputs, param_outputs, output_artifacts)
230     if outputs_dict:
231         template['outputs'] = outputs_dict
232 
233     # node selector
234     if processed_op.node_selector:
235         template['nodeSelector'] = processed_op.node_selector
236 
237     # tolerations
238     if processed_op.tolerations:
239         template['tolerations'] = processed_op.tolerations
240 
241     # affinity
242     if processed_op.affinity:
243         template['affinity'] = convert_k8s_obj_to_json(processed_op.affinity)
244 
245     # metadata
246     if processed_op.pod_annotations or processed_op.pod_labels:
247         template['metadata'] = {}
248         if processed_op.pod_annotations:
249             template['metadata']['annotations'] = processed_op.pod_annotations
250         if processed_op.pod_labels:
251             template['metadata']['labels'] = processed_op.pod_labels
252     # retries
253     if processed_op.num_retries:
254         template['retryStrategy'] = {'limit': processed_op.num_retries}
255 
256     # timeout
257     if processed_op.timeout:
258         template['activeDeadlineSeconds'] = processed_op.timeout
259 
260     # initContainers
261     if processed_op.init_containers:
262         template['initContainers'] = processed_op.init_containers
263 
264     # sidecars
265     if processed_op.sidecars:
266         template['sidecars'] = processed_op.sidecars
267 
268     # volumes
269     if processed_op.volumes:
270         template['volumes'] = [convert_k8s_obj_to_json(volume) for volume in processed_op.volumes]
271         template['volumes'].sort(key=lambda x: x['name'])
272 
273     # Display name
274     if processed_op.display_name:
275         template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/task_display_name'] = processed_op.display_name
276 
277     if isinstance(op, dsl.ContainerOp) and op._metadata:
278         template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_spec'] = json.dumps(op._metadata.to_dict(), sort_keys=True)
279 
280     if hasattr(op, '_component_ref'):
281         template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_ref'] = json.dumps(op._component_ref.to_dict(), sort_keys=True)
282 
283     if isinstance(op, dsl.ContainerOp) and op.execution_options:
284         if op.execution_options.caching_strategy.max_cache_staleness:
285             template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/max_cache_staleness'] = str(op.execution_options.caching_strategy.max_cache_staleness)
286 
287     return template
288
```
--- 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/compiler/_op_to_template.py b/sdk/python/kfp/compiler/_op_to_template.py
--- a/sdk/python/kfp/compiler/_op_to_template.py
+++ b/sdk/python/kfp/compiler/_op_to_template.py
@@ -176,6 +176,10 @@
 def _op_to_template(op: BaseOp):
     """Generate template given an operator inherited from BaseOp."""
 
+    # Display name
+    if op.display_name:
+        op.add_pod_annotation('pipelines.kubeflow.org/task_display_name', op.display_name)
+
     # NOTE in-place update to BaseOp
     # replace all PipelineParams with template var strings
     processed_op = _process_base_ops(op)
@@ -270,10 +274,6 @@
         template['volumes'] = [convert_k8s_obj_to_json(volume) for volume in processed_op.volumes]
         template['volumes'].sort(key=lambda x: x['name'])
 
-    # Display name
-    if processed_op.display_name:
-        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/task_display_name'] = processed_op.display_name
-
     if isinstance(op, dsl.ContainerOp) and op._metadata:
         template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_spec'] = json.dumps(op._metadata.to_dict(), sort_keys=True)

{"golden_diff": "diff --git a/sdk/python/kfp/compiler/_op_to_template.py b/sdk/python/kfp/compiler/_op_to_template.py\n--- a/sdk/python/kfp/compiler/_op_to_template.py\n+++ b/sdk/python/kfp/compiler/_op_to_template.py\n@@ -176,6 +176,10 @@\n def _op_to_template(op: BaseOp):\n     \"\"\"Generate template given an operator inherited from BaseOp.\"\"\"\n \n+    # Display name\n+    if op.display_name:\n+        op.add_pod_annotation('pipelines.kubeflow.org/task_display_name', op.display_name)\n+\n     # NOTE in-place update to BaseOp\n     # replace all PipelineParams with template var strings\n     processed_op = _process_base_ops(op)\n@@ -270,10 +274,6 @@\n         template['volumes'] = [convert_k8s_obj_to_json(volume) for volume in processed_op.volumes]\n         template['volumes'].sort(key=lambda x: x['name'])\n \n-    # Display name\n-    if processed_op.display_name:\n-        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/task_display_name'] = processed_op.display_name\n-\n     if isinstance(op, dsl.ContainerOp) and op._metadata:\n         template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_spec'] = json.dumps(op._metadata.to_dict(), sort_keys=True)\n", "issue": "SDK - Support placeholders in task display names\nFixes https://github.com/kubeflow/pipelines/issues/4145\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\nimport json\nimport re\nimport warnings\nimport yaml\nfrom collections import OrderedDict\nfrom typing import Union, List, Any, Callable, TypeVar, Dict\n\nfrom ._k8s_helper import convert_k8s_obj_to_json\nfrom .. import dsl\nfrom ..dsl._container_op import BaseOp\n\n\n# generics\nT = TypeVar('T')\n\n\ndef _process_obj(obj: Any, map_to_tmpl_var: dict):\n    \"\"\"Recursively sanitize and replace any PipelineParam (instances and serialized strings)\n    in the object with the corresponding template variables\n    (i.e. '{{inputs.parameters.<PipelineParam.full_name>}}').\n    \n    Args:\n      obj: any obj that may have PipelineParam\n      map_to_tmpl_var: a dict that maps an unsanitized pipeline\n                       params signature into a template var\n    \"\"\"\n    # serialized str might be unsanitized\n    if isinstance(obj, str):\n        # get signature\n        param_tuples = dsl.match_serialized_pipelineparam(obj)\n        if not param_tuples:\n            return obj\n        # replace all unsanitized signature with template var\n        for param_tuple in param_tuples:\n            obj = re.sub(param_tuple.pattern, map_to_tmpl_var[param_tuple.pattern], obj)\n\n    # list\n    if isinstance(obj, list):\n        return [_process_obj(item, map_to_tmpl_var) for item in obj]\n\n    # tuple\n    if isinstance(obj, tuple):\n        return tuple((_process_obj(item, map_to_tmpl_var) for item in obj))\n\n    # dict\n    if isinstance(obj, dict):\n        return {\n            _process_obj(key, map_to_tmpl_var): _process_obj(value, map_to_tmpl_var)\n            for key, value in obj.items()\n        }\n\n    # pipelineparam\n    if isinstance(obj, dsl.PipelineParam):\n        # if not found in unsanitized map, then likely to be sanitized\n        return map_to_tmpl_var.get(str(obj), '{{inputs.parameters.%s}}' % obj.full_name)\n\n    # k8s objects (generated from swaggercodegen)\n    if hasattr(obj, 'attribute_map') and isinstance(obj.attribute_map, dict):\n        # process everything inside recursively\n        for key in obj.attribute_map.keys():\n            setattr(obj, key, _process_obj(getattr(obj, key), map_to_tmpl_var))\n        # return json representation of the k8s obj\n        return convert_k8s_obj_to_json(obj)\n\n    # do nothing\n    return obj\n\n\ndef _process_base_ops(op: BaseOp):\n    \"\"\"Recursively go through the attrs listed in `attrs_with_pipelineparams`\n    and sanitize and replace pipeline params with template var string.\n\n    Returns a processed `BaseOp`.\n\n    NOTE this is an in-place update to `BaseOp`'s attributes (i.e. the ones\n    specified in `attrs_with_pipelineparams`, all `PipelineParam` are replaced\n    with the corresponding template variable strings).\n\n    Args:\n        op {BaseOp}: class that inherits from BaseOp\n\n    Returns:\n        BaseOp\n    \"\"\"\n\n    # map param's (unsanitized pattern or serialized str pattern) -> input param var str\n    map_to_tmpl_var = {\n        (param.pattern or str(param)): '{{inputs.parameters.%s}}' % param.full_name\n        for param in op.inputs\n    }\n\n    # process all attr with pipelineParams except inputs and outputs parameters\n    for key in op.attrs_with_pipelineparams:\n        setattr(op, key, _process_obj(getattr(op, key), map_to_tmpl_var))\n\n    return op\n\n\ndef _parameters_to_json(params: List[dsl.PipelineParam]):\n    \"\"\"Converts a list of PipelineParam into an argo `parameter` JSON obj.\"\"\"\n    _to_json = (lambda param: dict(name=param.full_name, value=param.value)\n                if param.value else dict(name=param.full_name))\n    params = [_to_json(param) for param in params]\n    # Sort to make the results deterministic.\n    params.sort(key=lambda x: x['name'])\n    return params\n\n\ndef _inputs_to_json(\n    inputs_params: List[dsl.PipelineParam],\n    input_artifact_paths: Dict[str, str] = None,\n    artifact_arguments: Dict[str, str] = None,\n) -> Dict[str, Dict]:\n    \"\"\"Converts a list of PipelineParam into an argo `inputs` JSON obj.\"\"\"\n    parameters = _parameters_to_json(inputs_params)\n\n    # Building the input artifacts section\n    artifacts = []\n    for name, path in (input_artifact_paths or {}).items():\n        artifact = {'name': name, 'path': path}\n        if name in artifact_arguments: # The arguments should be compiled as DAG task arguments, not template's default values, but in the current DSL-compiler implementation it's too hard to make that work when passing artifact references.\n            artifact['raw'] = {'data': str(artifact_arguments[name])}\n        artifacts.append(artifact)\n    artifacts.sort(key=lambda x: x['name']) #Stabilizing the input artifact ordering\n\n    inputs_dict = {}\n    if parameters:\n        inputs_dict['parameters'] = parameters\n    if artifacts:\n        inputs_dict['artifacts'] = artifacts\n    return inputs_dict\n\n\ndef _outputs_to_json(op: BaseOp,\n                     outputs: Dict[str, dsl.PipelineParam],\n                     param_outputs: Dict[str, str],\n                     output_artifacts: List[dict]):\n    \"\"\"Creates an argo `outputs` JSON obj.\"\"\"\n    if isinstance(op, dsl.ResourceOp):\n        value_from_key = \"jsonPath\"\n    else:\n        value_from_key = \"path\"\n    output_parameters = []\n    for param in set(outputs.values()):  # set() dedupes output references\n        output_parameters.append({\n            'name': param.full_name,\n            'valueFrom': {\n                value_from_key: param_outputs[param.name]\n            }\n        })\n    output_parameters.sort(key=lambda x: x['name'])\n    ret = {}\n    if output_parameters:\n        ret['parameters'] = output_parameters\n    if output_artifacts:\n        ret['artifacts'] = output_artifacts\n\n    return ret\n\n\n# TODO: generate argo python classes from swagger and use convert_k8s_obj_to_json??\ndef _op_to_template(op: BaseOp):\n    \"\"\"Generate template given an operator inherited from BaseOp.\"\"\"\n\n    # NOTE in-place update to BaseOp\n    # replace all PipelineParams with template var strings\n    processed_op = _process_base_ops(op)\n\n    if isinstance(op, dsl.ContainerOp):\n        # default output artifacts\n        output_artifact_paths = OrderedDict(op.output_artifact_paths)\n        # This should have been as easy as output_artifact_paths.update(op.file_outputs), but the _outputs_to_json function changes the output names and we must do the same here, so that the names are the same\n        output_artifact_paths.update(sorted(((param.full_name, processed_op.file_outputs[param.name]) for param in processed_op.outputs.values()), key=lambda x: x[0]))\n\n        output_artifacts = [\n            {'name': name, 'path': path}\n            for name, path in output_artifact_paths.items()\n        ]\n\n        # workflow template\n        template = {\n            'name': processed_op.name,\n            'container': convert_k8s_obj_to_json(\n                processed_op.container\n            )\n        }\n    elif isinstance(op, dsl.ResourceOp):\n        # no output artifacts\n        output_artifacts = []\n\n        # workflow template\n        processed_op.resource[\"manifest\"] = yaml.dump(\n            convert_k8s_obj_to_json(processed_op.k8s_resource),\n            default_flow_style=False\n        )\n        template = {\n            'name': processed_op.name,\n            'resource': convert_k8s_obj_to_json(\n                processed_op.resource\n            )\n        }\n\n    # inputs\n    input_artifact_paths = processed_op.input_artifact_paths if isinstance(processed_op, dsl.ContainerOp) else None\n    artifact_arguments = processed_op.artifact_arguments if isinstance(processed_op, dsl.ContainerOp) else None\n    inputs = _inputs_to_json(processed_op.inputs, input_artifact_paths, artifact_arguments)\n    if inputs:\n        template['inputs'] = inputs\n\n    # outputs\n    if isinstance(op, dsl.ContainerOp):\n        param_outputs = processed_op.file_outputs\n    elif isinstance(op, dsl.ResourceOp):\n        param_outputs = processed_op.attribute_outputs\n    outputs_dict = _outputs_to_json(op, processed_op.outputs, param_outputs, output_artifacts)\n    if outputs_dict:\n        template['outputs'] = outputs_dict\n\n    # node selector\n    if processed_op.node_selector:\n        template['nodeSelector'] = processed_op.node_selector\n\n    # tolerations\n    if processed_op.tolerations:\n        template['tolerations'] = processed_op.tolerations\n\n    # affinity\n    if processed_op.affinity:\n        template['affinity'] = convert_k8s_obj_to_json(processed_op.affinity)\n\n    # metadata\n    if processed_op.pod_annotations or processed_op.pod_labels:\n        template['metadata'] = {}\n        if processed_op.pod_annotations:\n            template['metadata']['annotations'] = processed_op.pod_annotations\n        if processed_op.pod_labels:\n            template['metadata']['labels'] = processed_op.pod_labels\n    # retries\n    if processed_op.num_retries:\n        template['retryStrategy'] = {'limit': processed_op.num_retries}\n\n    # timeout\n    if processed_op.timeout:\n        template['activeDeadlineSeconds'] = processed_op.timeout\n\n    # initContainers\n    if processed_op.init_containers:\n        template['initContainers'] = processed_op.init_containers\n\n    # sidecars\n    if processed_op.sidecars:\n        template['sidecars'] = processed_op.sidecars\n\n    # volumes\n    if processed_op.volumes:\n        template['volumes'] = [convert_k8s_obj_to_json(volume) for volume in processed_op.volumes]\n        template['volumes'].sort(key=lambda x: x['name'])\n\n    # Display name\n    if processed_op.display_name:\n        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/task_display_name'] = processed_op.display_name\n\n    if isinstance(op, dsl.ContainerOp) and op._metadata:\n        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_spec'] = json.dumps(op._metadata.to_dict(), sort_keys=True)\n\n    if hasattr(op, '_component_ref'):\n        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_ref'] = json.dumps(op._component_ref.to_dict(), sort_keys=True)\n\n    if isinstance(op, dsl.ContainerOp) and op.execution_options:\n        if op.execution_options.caching_strategy.max_cache_staleness:\n            template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/max_cache_staleness'] = str(op.execution_options.caching_strategy.max_cache_staleness)\n\n    return template\n", "path": "sdk/python/kfp/compiler/_op_to_template.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\nimport json\nimport re\nimport warnings\nimport yaml\nfrom collections import OrderedDict\nfrom typing import Union, List, Any, Callable, TypeVar, Dict\n\nfrom ._k8s_helper import convert_k8s_obj_to_json\nfrom .. import dsl\nfrom ..dsl._container_op import BaseOp\n\n\n# generics\nT = TypeVar('T')\n\n\ndef _process_obj(obj: Any, map_to_tmpl_var: dict):\n    \"\"\"Recursively sanitize and replace any PipelineParam (instances and serialized strings)\n    in the object with the corresponding template variables\n    (i.e. '{{inputs.parameters.<PipelineParam.full_name>}}').\n    \n    Args:\n      obj: any obj that may have PipelineParam\n      map_to_tmpl_var: a dict that maps an unsanitized pipeline\n                       params signature into a template var\n    \"\"\"\n    # serialized str might be unsanitized\n    if isinstance(obj, str):\n        # get signature\n        param_tuples = dsl.match_serialized_pipelineparam(obj)\n        if not param_tuples:\n            return obj\n        # replace all unsanitized signature with template var\n        for param_tuple in param_tuples:\n            obj = re.sub(param_tuple.pattern, map_to_tmpl_var[param_tuple.pattern], obj)\n\n    # list\n    if isinstance(obj, list):\n        return [_process_obj(item, map_to_tmpl_var) for item in obj]\n\n    # tuple\n    if isinstance(obj, tuple):\n        return tuple((_process_obj(item, map_to_tmpl_var) for item in obj))\n\n    # dict\n    if isinstance(obj, dict):\n        return {\n            _process_obj(key, map_to_tmpl_var): _process_obj(value, map_to_tmpl_var)\n            for key, value in obj.items()\n        }\n\n    # pipelineparam\n    if isinstance(obj, dsl.PipelineParam):\n        # if not found in unsanitized map, then likely to be sanitized\n        return map_to_tmpl_var.get(str(obj), '{{inputs.parameters.%s}}' % obj.full_name)\n\n    # k8s objects (generated from swaggercodegen)\n    if hasattr(obj, 'attribute_map') and isinstance(obj.attribute_map, dict):\n        # process everything inside recursively\n        for key in obj.attribute_map.keys():\n            setattr(obj, key, _process_obj(getattr(obj, key), map_to_tmpl_var))\n        # return json representation of the k8s obj\n        return convert_k8s_obj_to_json(obj)\n\n    # do nothing\n    return obj\n\n\ndef _process_base_ops(op: BaseOp):\n    \"\"\"Recursively go through the attrs listed in `attrs_with_pipelineparams`\n    and sanitize and replace pipeline params with template var string.\n\n    Returns a processed `BaseOp`.\n\n    NOTE this is an in-place update to `BaseOp`'s attributes (i.e. the ones\n    specified in `attrs_with_pipelineparams`, all `PipelineParam` are replaced\n    with the corresponding template variable strings).\n\n    Args:\n        op {BaseOp}: class that inherits from BaseOp\n\n    Returns:\n        BaseOp\n    \"\"\"\n\n    # map param's (unsanitized pattern or serialized str pattern) -> input param var str\n    map_to_tmpl_var = {\n        (param.pattern or str(param)): '{{inputs.parameters.%s}}' % param.full_name\n        for param in op.inputs\n    }\n\n    # process all attr with pipelineParams except inputs and outputs parameters\n    for key in op.attrs_with_pipelineparams:\n        setattr(op, key, _process_obj(getattr(op, key), map_to_tmpl_var))\n\n    return op\n\n\ndef _parameters_to_json(params: List[dsl.PipelineParam]):\n    \"\"\"Converts a list of PipelineParam into an argo `parameter` JSON obj.\"\"\"\n    _to_json = (lambda param: dict(name=param.full_name, value=param.value)\n                if param.value else dict(name=param.full_name))\n    params = [_to_json(param) for param in params]\n    # Sort to make the results deterministic.\n    params.sort(key=lambda x: x['name'])\n    return params\n\n\ndef _inputs_to_json(\n    inputs_params: List[dsl.PipelineParam],\n    input_artifact_paths: Dict[str, str] = None,\n    artifact_arguments: Dict[str, str] = None,\n) -> Dict[str, Dict]:\n    \"\"\"Converts a list of PipelineParam into an argo `inputs` JSON obj.\"\"\"\n    parameters = _parameters_to_json(inputs_params)\n\n    # Building the input artifacts section\n    artifacts = []\n    for name, path in (input_artifact_paths or {}).items():\n        artifact = {'name': name, 'path': path}\n        if name in artifact_arguments: # The arguments should be compiled as DAG task arguments, not template's default values, but in the current DSL-compiler implementation it's too hard to make that work when passing artifact references.\n            artifact['raw'] = {'data': str(artifact_arguments[name])}\n        artifacts.append(artifact)\n    artifacts.sort(key=lambda x: x['name']) #Stabilizing the input artifact ordering\n\n    inputs_dict = {}\n    if parameters:\n        inputs_dict['parameters'] = parameters\n    if artifacts:\n        inputs_dict['artifacts'] = artifacts\n    return inputs_dict\n\n\ndef _outputs_to_json(op: BaseOp,\n                     outputs: Dict[str, dsl.PipelineParam],\n                     param_outputs: Dict[str, str],\n                     output_artifacts: List[dict]):\n    \"\"\"Creates an argo `outputs` JSON obj.\"\"\"\n    if isinstance(op, dsl.ResourceOp):\n        value_from_key = \"jsonPath\"\n    else:\n        value_from_key = \"path\"\n    output_parameters = []\n    for param in set(outputs.values()):  # set() dedupes output references\n        output_parameters.append({\n            'name': param.full_name,\n            'valueFrom': {\n                value_from_key: param_outputs[param.name]\n            }\n        })\n    output_parameters.sort(key=lambda x: x['name'])\n    ret = {}\n    if output_parameters:\n        ret['parameters'] = output_parameters\n    if output_artifacts:\n        ret['artifacts'] = output_artifacts\n\n    return ret\n\n\n# TODO: generate argo python classes from swagger and use convert_k8s_obj_to_json??\ndef _op_to_template(op: BaseOp):\n    \"\"\"Generate template given an operator inherited from BaseOp.\"\"\"\n\n    # Display name\n    if op.display_name:\n        op.add_pod_annotation('pipelines.kubeflow.org/task_display_name', op.display_name)\n\n    # NOTE in-place update to BaseOp\n    # replace all PipelineParams with template var strings\n    processed_op = _process_base_ops(op)\n\n    if isinstance(op, dsl.ContainerOp):\n        # default output artifacts\n        output_artifact_paths = OrderedDict(op.output_artifact_paths)\n        # This should have been as easy as output_artifact_paths.update(op.file_outputs), but the _outputs_to_json function changes the output names and we must do the same here, so that the names are the same\n        output_artifact_paths.update(sorted(((param.full_name, processed_op.file_outputs[param.name]) for param in processed_op.outputs.values()), key=lambda x: x[0]))\n\n        output_artifacts = [\n            {'name': name, 'path': path}\n            for name, path in output_artifact_paths.items()\n        ]\n\n        # workflow template\n        template = {\n            'name': processed_op.name,\n            'container': convert_k8s_obj_to_json(\n                processed_op.container\n            )\n        }\n    elif isinstance(op, dsl.ResourceOp):\n        # no output artifacts\n        output_artifacts = []\n\n        # workflow template\n        processed_op.resource[\"manifest\"] = yaml.dump(\n            convert_k8s_obj_to_json(processed_op.k8s_resource),\n            default_flow_style=False\n        )\n        template = {\n            'name': processed_op.name,\n            'resource': convert_k8s_obj_to_json(\n                processed_op.resource\n            )\n        }\n\n    # inputs\n    input_artifact_paths = processed_op.input_artifact_paths if isinstance(processed_op, dsl.ContainerOp) else None\n    artifact_arguments = processed_op.artifact_arguments if isinstance(processed_op, dsl.ContainerOp) else None\n    inputs = _inputs_to_json(processed_op.inputs, input_artifact_paths, artifact_arguments)\n    if inputs:\n        template['inputs'] = inputs\n\n    # outputs\n    if isinstance(op, dsl.ContainerOp):\n        param_outputs = processed_op.file_outputs\n    elif isinstance(op, dsl.ResourceOp):\n        param_outputs = processed_op.attribute_outputs\n    outputs_dict = _outputs_to_json(op, processed_op.outputs, param_outputs, output_artifacts)\n    if outputs_dict:\n        template['outputs'] = outputs_dict\n\n    # node selector\n    if processed_op.node_selector:\n        template['nodeSelector'] = processed_op.node_selector\n\n    # tolerations\n    if processed_op.tolerations:\n        template['tolerations'] = processed_op.tolerations\n\n    # affinity\n    if processed_op.affinity:\n        template['affinity'] = convert_k8s_obj_to_json(processed_op.affinity)\n\n    # metadata\n    if processed_op.pod_annotations or processed_op.pod_labels:\n        template['metadata'] = {}\n        if processed_op.pod_annotations:\n            template['metadata']['annotations'] = processed_op.pod_annotations\n        if processed_op.pod_labels:\n            template['metadata']['labels'] = processed_op.pod_labels\n    # retries\n    if processed_op.num_retries:\n        template['retryStrategy'] = {'limit': processed_op.num_retries}\n\n    # timeout\n    if processed_op.timeout:\n        template['activeDeadlineSeconds'] = processed_op.timeout\n\n    # initContainers\n    if processed_op.init_containers:\n        template['initContainers'] = processed_op.init_containers\n\n    # sidecars\n    if processed_op.sidecars:\n        template['sidecars'] = processed_op.sidecars\n\n    # volumes\n    if processed_op.volumes:\n        template['volumes'] = [convert_k8s_obj_to_json(volume) for volume in processed_op.volumes]\n        template['volumes'].sort(key=lambda x: x['name'])\n\n    if isinstance(op, dsl.ContainerOp) and op._metadata:\n        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_spec'] = json.dumps(op._metadata.to_dict(), sort_keys=True)\n\n    if hasattr(op, '_component_ref'):\n        template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_ref'] = json.dumps(op._component_ref.to_dict(), sort_keys=True)\n\n    if isinstance(op, dsl.ContainerOp) and op.execution_options:\n        if op.execution_options.caching_strategy.max_cache_staleness:\n            template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/max_cache_staleness'] = str(op.execution_options.caching_strategy.max_cache_staleness)\n\n    return template\n", "path": "sdk/python/kfp/compiler/_op_to_template.py"}]}