extract the pipelineparam deserialize function

sdk/python/kfp/dsl/_container_op.py

@@ -15,6 +15,7 @@
 
 from . import _pipeline
 from . import _pipeline_param
+from ._pipeline_param import _extract_pipelineparams
 import re
 from typing import Dict
 
@@ -64,13 +65,8 @@ def __init__(self, name: str, image: str, command: str=None, arguments: str=None
     self.pod_labels = {}
     self.num_retries = 0
 
-    matches = []
-    for arg in (command or []) + (arguments or []):
-      match = re.findall(r'{{pipelineparam:op=([\w\s_-]*);name=([\w\s_-]+);value=(.*?)}}', str(arg))
-      matches += match
+    self.argument_inputs = _extract_pipelineparams([str(arg) for arg in (command or []) + (arguments or [])])
 
-    self.argument_inputs = [_pipeline_param.PipelineParam(x[1], x[0], x[2])
-                            for x in list(set(matches))]
     self.file_outputs = file_outputs
     self.dependent_op_names = []
 

sdk/python/kfp/dsl/_pipeline_param.py

@@ -21,6 +21,21 @@
 # For now, this identifies a condition with only "==" operator supported.
 ConditionOperator = namedtuple('ConditionOperator', 'operator operand1 operand2')
 
+def _extract_pipelineparams(payloads: str or list[str]):
+  """_extract_pipelineparam extract a list of PipelineParam instances from the payload string.
+  Note: this function removes all duplicate matches.
+
+  Args:
+    payload (str or list[str]): a string/a list of strings that contains serialized pipelineparams
+  Return:
+    List[PipelineParam]
+  """
+  if isinstance(payloads, str):
+    payloads = [payloads]
+  matches = []
+  for payload in payloads:
+    matches += re.findall(r'{{pipelineparam:op=([\w\s_-]*);name=([\w\s_-]+);value=(.*?)}}', payload)
+  return [PipelineParam(x[1], x[0], x[2]) for x in list(set(matches))]
 
 class PipelineParam(object):
   """Representing a future value that is passed between pipeline components.

sdk/python/tests/dsl/pipeline_param_tests.py

@@ -14,6 +14,7 @@
 
 
 from kfp.dsl import PipelineParam
+from kfp.dsl._pipeline_param import _extract_pipelineparams
 import unittest
 
 
@@ -35,3 +36,17 @@ def test_str_repr(self):
 
     p = PipelineParam(name='param3', value='value3')
     self.assertEqual('{{pipelineparam:op=;name=param3;value=value3}}', str(p))
+
+  def test_extract_pipelineparam(self):
+    """Test _extract_pipeleineparam."""
+
+    p1 = PipelineParam(name='param1', op_name='op1')
+    p2 = PipelineParam(name='param2')
+    p3 = PipelineParam(name='param3', value='value3')
+    stuff_chars = ' between '
+    payload = str(p1) + stuff_chars + str(p2) + stuff_chars + str(p3)
+    params = _extract_pipelineparams(payload)
+    self.assertListEqual([p1, p2, p3], params)
+    payload = [str(p1) + stuff_chars + str(p2), str(p2) + stuff_chars + str(p3)]
+    params = _extract_pipelineparams(payload)
+    self.assertListEqual([p1, p2, p3], params)