Add support to parse pipeline param in json string (#841)

* add support to parse pipeline param in json string

* update license

sdk/python/kfp_tekton/compiler/compiler.py

@@ -383,17 +383,43 @@ def _group_to_dag_template(self, group, inputs, outputs, dependencies, pipeline_
         }]
       else:
         # Need to sanitize the dict keys for consistency.
+        def process_pipelineparam(s):
+          if "{{pipelineparam" in s:
+            pipe_params = re.findall(r"{{pipelineparam:op=([^ \t\n,]*);name=([^ \t\n,]*)}}", s)
+            for pipe_param in pipe_params:
+              if pipe_param[0] == '':
+                s = s.replace("{{pipelineparam:op=%s;name=%s}}" % pipe_param, '$(params.%s)' % pipe_param[1])
+              else:
+                param_name = sanitize_k8s_name(pipe_param[1])
+                s = s.replace("{{pipelineparam:op=%s;name=%s}}" % pipe_param, '$(tasks.%s.results.%s)' % (
+                  sanitize_k8s_name(pipe_param[0]),
+                  param_name))
+          return s
         loop_arg_value = sub_group.loop_args.to_list_for_task_yaml()
         loop_args_str_value = ''
         sanitized_tasks = []
         if isinstance(loop_arg_value[0], dict):
           for argument_set in loop_arg_value:
             c_dict = {}
             for k, v in argument_set.items():
+              if isinstance(v, dsl.PipelineParam):
+                if v.op_name is None:
+                  v = '$(params.%s)' % v.name
+                else:
+                  param_name = sanitize_k8s_name(v.name)
+                  v = '$(tasks.%s.results.%s)' % (
+                    sanitize_k8s_name(v.op_name),
+                    param_name)
+              else:
+                if isinstance(v, str):
+                  v = process_pipelineparam(v)
               c_dict[sanitize_k8s_name(k, True)] = v
             sanitized_tasks.append(c_dict)
           loop_args_str_value = json.dumps(sanitized_tasks, sort_keys=True)
         else:
+          for i, value in enumerate(loop_arg_value):
+            if isinstance(value, str):
+              loop_arg_value[i] = process_pipelineparam(value)
           loop_args_str_value = json.dumps(loop_arg_value)
 
         self.loops_pipeline[group_name]['spec']['params'] = [{

sdk/python/tests/compiler/compiler_tests.py

@@ -84,6 +84,13 @@ def test_condition_custom_task_workflow(self):
     from .testdata.condition_custom_task import flipcoin_pipeline
     self._test_pipeline_workflow(flipcoin_pipeline, 'condition_custom_task.yaml', skip_noninlined=True)
 
+  def test_loop_with_params_in_json_workflow(self):
+    """
+    Test compiling a loop workflow with pipeline params in json
+    """
+    from .testdata.loop_with_params_in_json import parallelfor_pipeline_param_in_items_resolving
+    self._test_pipeline_workflow(parallelfor_pipeline_param_in_items_resolving, 'loop_with_params_in_json.yaml', skip_noninlined=True)
+
   def test_condition_dependency(self):
     """
     Test dependency on Tekton conditional task.

sdk/python/tests/compiler/testdata/loop_literal_separator.yaml

@@ -1,4 +1,4 @@
-# Copyright 2022 kubeflow.org
+# Copyright 2021 kubeflow.org
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

sdk/python/tests/compiler/testdata/loop_literal_separator_noninlined.yaml

@@ -1,4 +1,4 @@
-# Copyright 2022 kubeflow.org
+# Copyright 2021 kubeflow.org
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -38,11 +38,11 @@ metadata:
       "{\"name\": \"my-in-coop1\", \"outputs\": [], \"version\": \"my-in-coop1@sha256=42f0c6a60c7e435441b8afaeb382a771a9741fe3aabb203748fdbd72b25f1628\"}",
       "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
       "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
-      ""}}, "params": [{"name": "loop-item-param-2", "type": "string"},
-      {"name": "my_pipe_param", "type": "string"}], "steps": [{"args": ["set -e\necho
-      op1 \"$0\" \"$1\"\n", "$(inputs.params.loop-item-param-2)", "$(inputs.params.my_pipe_param)"],
-      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
-      "timeout": "525600m"}]}}}]'
+      ""}}, "params": [{"name": "loop-item-param-2", "type": "string"}, {"name": "my_pipe_param",
+      "type": "string"}], "steps": [{"args": ["set -e\necho op1 \"$0\" \"$1\"\n",
+      "$(inputs.params.loop-item-param-2)", "$(inputs.params.my_pipe_param)"], "command":
+      ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]}, "timeout":
+      "525600m"}]}}}]'
 spec:
   params:
   - name: my_pipe_param

sdk/python/tests/compiler/testdata/loop_with_params_in_json.py

@@ -0,0 +1,64 @@
+# Copyright 2022 kubeflow.org
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import kfp
+from kfp.components import func_to_container_op
+
+
+@func_to_container_op
+def produce_message(fname1: str) -> str:
+    return "My name is %s" % fname1
+
+
+@func_to_container_op
+def consume(param1):
+    print(param1)
+
+
+@kfp.dsl.pipeline()
+def parallelfor_pipeline_param_in_items_resolving(fname1: str, fname2: str):
+
+    simple_list = ["My name is %s" % fname1, "My name is %s" % fname2]
+
+    list_of_dict = [{
+        "first_name": fname1,
+        "message": "My name is %s" % fname1
+    }, {
+        "first_name": fname2,
+        "message": "My name is %s" % fname2
+    }]
+
+    list_of_complex_dict = [{
+        "first_name": fname1,
+        "message": produce_message(fname1).output
+    }, {
+        "first_name": fname2,
+        "message": produce_message(fname2).output
+    }]
+
+    with kfp.dsl.ParallelFor(simple_list) as loop_item:
+        consume(loop_item)
+
+    with kfp.dsl.ParallelFor(list_of_dict) as loop_item2:
+        consume(loop_item2.first_name)
+        consume(loop_item2.message)
+
+    with kfp.dsl.ParallelFor(list_of_complex_dict) as loop_item:
+        consume(loop_item.first_name)
+        consume(loop_item.message)
+
+
+if __name__ == '__main__':
+    from kfp_tekton.compiler import TektonCompiler
+    TektonCompiler().compile(parallelfor_pipeline_param_in_items_resolving, __file__.replace('.py', '.yaml'))