Add support for pipeline parameters

elyra/elyra_app.py

@@ -37,6 +37,7 @@
 from elyra.pipeline.handlers import PipelineComponentHandler
 from elyra.pipeline.handlers import PipelineComponentPropertiesHandler
 from elyra.pipeline.handlers import PipelineExportHandler
+from elyra.pipeline.handlers import PipelineParametersHandler
 from elyra.pipeline.handlers import PipelinePropertiesHandler
 from elyra.pipeline.handlers import PipelineRuntimeTypesHandler
 from elyra.pipeline.handlers import PipelineSchedulerHandler
@@ -109,6 +110,7 @@ def initialize_handlers(self):
                 ),
                 (f"/{self.name}/pipeline/export", PipelineExportHandler),
                 (f"/{self.name}/pipeline/{processor_regex}/properties", PipelinePropertiesHandler),
+                (f"/{self.name}/pipeline/{processor_regex}/parameters", PipelineParametersHandler),
                 (f"/{self.name}/pipeline/runtimes/types", PipelineRuntimeTypesHandler),
                 (f"/{self.name}/pipeline/schedule", PipelineSchedulerHandler),
                 (f"/{self.name}/pipeline/validate", PipelineValidationHandler),

elyra/pipeline/component.py

@@ -17,6 +17,7 @@
 
 from abc import abstractmethod
 from dataclasses import dataclass
+from dataclasses import field
 from importlib import import_module
 import json
 from logging import Logger
@@ -307,6 +308,11 @@ def determine_type_information(self, parsed_type: str) -> "ParameterTypeInfo":
             else:  # Let this be undetermined. Callers should check for this status and adjust
                 data_type_info = ParameterTypeInfo(parsed_data=parsed_type_lowered, undetermined=True)
 
+        from elyra.pipeline.processor import PipelineProcessorManager  # placed here to avoid circular reference
+
+        if PipelineProcessorManager.instance().supports_parameters(runtime_type=self.component_platform):
+            data_type_info.allowed_input_types.append("parameter")
+
         return data_type_info
 
 
@@ -328,7 +334,7 @@ class ParameterTypeInfo:
 
     parsed_data: str
     json_data_type: Optional[str] = "string"
-    allowed_input_types: Optional[List[str]] = None
+    allowed_input_types: Optional[List[str]] = field(default_factory=lambda: ["inputvalue", "inputpath", "file"])
     default_value: Optional[Any] = ""
     required: Optional[bool] = True
     undetermined: Optional[bool] = False

elyra/pipeline/component_parameter.py

@@ -41,6 +41,7 @@
 from elyra.pipeline.pipeline_constants import KUBERNETES_SHARED_MEM_SIZE
 from elyra.pipeline.pipeline_constants import KUBERNETES_TOLERATIONS
 from elyra.pipeline.pipeline_constants import MOUNTED_VOLUMES
+from elyra.pipeline.pipeline_constants import PIPELINE_PARAMETERS
 from elyra.util.kubernetes import is_valid_annotation_key
 from elyra.util.kubernetes import is_valid_annotation_value
 from elyra.util.kubernetes import is_valid_kubernetes_key
@@ -193,10 +194,10 @@ def get_classes_for_component_type(cls, component_type: str, runtime_type: Optio
         processor_props = set()
         for processor in PipelineProcessorManager.instance().get_all_processors():
             props = getattr(processor, "supported_properties", set())
-            if processor.type.name == runtime_type and props:
-                processor_props = props  # correct processor is found, and it explicitly specifies its properties
-                break
-            processor_props.update(props)
+            if not runtime_type or processor.type.name == runtime_type:
+                processor_props.update(props)
+            # if processor.type.name == runtime_type and props:
+            #    break  # TODO we don't want to break this soon, considering the 1:M notion of runtime type processors
 
         all_subclasses = set()
         for sc in cls.all_subclasses():
@@ -969,8 +970,8 @@ def __init__(
         name: str,
         json_data_type: str,
         description: str,
+        allowed_input_types: List[Optional[str]],
         value: Optional[Any] = "",
-        allowed_input_types: Optional[List[Optional[str]]] = None,
         required: Optional[bool] = False,
         allow_no_options: Optional[bool] = False,
         items: Optional[List[str]] = None,
@@ -1018,11 +1019,7 @@ def __init__(
         self._value = value
 
         self._description = description
-
-        if not allowed_input_types:
-            allowed_input_types = ["inputvalue", "inputpath", "file"]
         self._allowed_input_types = allowed_input_types
-
         self._items = items or []
 
         # Check description for information about 'required' parameter
@@ -1161,3 +1158,70 @@ class InputTypeDescriptionMap(Enum):
     file = "Please select a file to use as input:"
     inputpath = "Please select an output from a parent:"
     outputpath = None  # outputs are read-only and don't require a description
+    parameter = "Please select a parameter to use as input:"
+
+
+class PipelineParameter(ElyraPropertyListItem):
+    """TODO"""
+
+    property_id = PIPELINE_PARAMETERS
+    property_display_name = "Pipeline Parameters"
+    property_description = """TODO"""  # TODO
+    property_attributes = [
+        ListItemPropertyAttribute(
+            attribute_id="name",
+            display_name="Parameter Name",
+            input_type="string",
+            hidden=False,
+            required=True,
+            use_in_key=True,
+        ),
+        ListItemPropertyAttribute(
+            attribute_id="value",
+            display_name="Default Value",
+            placeholder="default_val",
+            input_type="string",
+            hidden=False,
+            required=False,  # TODO?
+            use_in_key=False,
+        ),
+        ListItemPropertyAttribute(
+            attribute_id="input_type",
+            display_name="Type",
+            placeholder="String",
+            input_type="string",
+            enum=["String"],
+            hidden=False,
+            required=True,
+            use_in_key=False,
+        ),
+        ListItemPropertyAttribute(
+            attribute_id="required",
+            display_name="Required",
+            input_type="boolean",
+            hidden=False,
+            required=True,
+            use_in_key=False,
+        ),
+    ]
+
+    def __init__(self, name, value, input_type, required, **kwargs):
+        self.name = name
+        self.value = value
+        self.input_type = input_type
+        self.required = required
+
+    def get_value_for_dict_entry(self) -> Union[str, Dict[str, Any]]:
+        """
+        Returns the value to be used when constructing a dict from a list of ElyraPropertyListItem.
+        A PipelineParameter dict entry will be of the form: {self.name: self.value}
+        """
+        return self.value
+
+    def get_all_validation_errors(self) -> List[str]:
+        # TODO Ensure param name adheres to python variable naming rules
+        pass
+
+    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
+        # TODO this path may or may not be reached depending on how processing is implemented
+        pass

elyra/pipeline/handlers.py

@@ -31,6 +31,7 @@
 from elyra.pipeline.component import Component
 from elyra.pipeline.component_catalog import ComponentCache
 from elyra.pipeline.component_catalog import RefreshInProgressError
+from elyra.pipeline.component_parameter import PipelineParameter
 from elyra.pipeline.parser import PipelineParser
 from elyra.pipeline.pipeline_definition import PipelineDefinition
 from elyra.pipeline.processor import PipelineProcessorManager
@@ -200,24 +201,44 @@ class PipelinePropertiesHandler(HttpErrorMixin, APIHandler):
 
     @web.authenticated
     async def get(self, runtime_type):
-        self.log.debug(f"Retrieving pipeline components for runtime type: {runtime_type}")
+        self.log.debug(f"Retrieving pipeline properties for runtime type: {runtime_type}")
 
         runtime_processor_type = get_runtime_processor_type(runtime_type, self.log, self.request.path)
         if not runtime_processor_type:
             raise web.HTTPError(400, f"Invalid runtime type '{runtime_type}'")
 
         # Get pipeline properties json
-        pipeline_properties_json = PipelineDefinition.get_canvas_properties_from_template(
-            package_name="templates/pipeline",
-            template_name="pipeline_properties_template.jinja2",
-            runtime_type=runtime_processor_type.name,
-        )
+        pipeline_properties_json = PipelineDefinition.get_pipeline_properties(runtime_type=runtime_processor_type.name)
 
         self.set_status(200)
         self.set_header("Content-Type", "application/json")
         await self.finish(pipeline_properties_json)
 
 
+class PipelineParametersHandler(HttpErrorMixin, APIHandler):
+    """Handler to expose method calls to retrieve pipeline parameters"""
+
+    @web.authenticated
+    async def get(self, runtime_type):
+        self.log.debug(f"Retrieving pipeline parameters for runtime type: {runtime_type}")
+
+        runtime_processor_type = get_runtime_processor_type(runtime_type, self.log, self.request.path)
+        if not runtime_processor_type:
+            raise web.HTTPError(400, f"Invalid runtime type '{runtime_type}'")
+
+        if PipelineProcessorManager.instance().supports_parameters(runtime_type=runtime_processor_type):
+            # Get pipeline parameters json
+            pipeline_properties_json = PipelineParameter.get_schema()
+            self.set_status(200)
+            self.set_header("Content-Type", "application/json")
+            await self.finish(pipeline_properties_json)
+        else:
+            # Pipeline parameters are not supported, no content to return
+            self.set_status(204)
+            processor_name = RuntimeProcessorType.get_instance_by_name(runtime_type)
+            await self.finish(f"{processor_name} does not support pipeline parameters.")
+
+
 class PipelineComponentPropertiesHandler(HttpErrorMixin, APIHandler):
     """Handler to expose method calls to retrieve pipeline component_id properties"""
 

elyra/pipeline/kfp/component_parser_kfp.py

@@ -202,7 +202,6 @@ def determine_type_information(self, parsed_type: str) -> SimpleNamespace:
         """
         data_type_info = super().determine_type_information(parsed_type)
 
-        data_type_info.allowed_input_types = []
         # By default, original input type (determined by parent) is stored as the `json_data_type`
         # and then overridden with Kubeflow Pipeline's meta-type
         if data_type_info.undetermined:

elyra/pipeline/kfp/processor_kfp.py

@@ -853,6 +853,11 @@ def supported_properties(self) -> Set[str]:
             pipeline_constants.KUBERNETES_SHARED_MEM_SIZE,
         }
 
+    @property
+    def supports_pipeline_parameters(self) -> bool:
+        """KfpPipelineProcessor does support pipeline parameters."""
+        return True
+
 
 class KfpPipelineProcessorResponse(RuntimePipelineProcessorResponse):
     _type = RuntimeProcessorType.KUBEFLOW_PIPELINES

elyra/pipeline/parser.py

@@ -61,7 +61,8 @@ def parse(self, pipeline_json: Dict) -> Pipeline:
             runtime_config=runtime_config,
             source=source,
             description=description,
-            pipeline_properties=primary_pipeline.get_pipeline_default_properties(),
+            pipeline_properties=primary_pipeline.pipeline_default_properties,
+            pipeline_parameters=primary_pipeline.pipeline_parameters,
         )
 
         nodes = primary_pipeline.nodes

elyra/pipeline/pipeline.py

@@ -22,7 +22,7 @@
 from typing import List
 from typing import Optional
 
-from elyra.pipeline.component_parameter import ElyraPropertyList
+from elyra.pipeline.component_parameter import ElyraPropertyList, PipelineParameter
 from elyra.pipeline.component_parameter import EnvironmentVariable
 from elyra.pipeline.pipeline_constants import ENV_VARIABLES
 from elyra.pipeline.pipeline_constants import RUNTIME_IMAGE
@@ -342,6 +342,7 @@ def __init__(
         source: Optional[str] = None,
         description: Optional[str] = None,
         pipeline_properties: Optional[Dict[str, Any]] = None,
+        pipeline_parameters: ElyraPropertyList[PipelineParameter] = None,
     ):
         """
         :param id: Generated UUID, 128 bit number used as a unique identifier
@@ -352,6 +353,7 @@ def __init__(
         :param source: The pipeline source, e.g. a pipeline file or a notebook.
         :param description: Pipeline description
         :param pipeline_properties: Key/value pairs representing the properties of this pipeline
+        :param pipeline_parameters: an ElyraPropertyList of the pipeline parameters
         """
 
         if not name:
@@ -366,6 +368,7 @@ def __init__(
         self._runtime = runtime
         self._runtime_config = runtime_config
         self._pipeline_properties = pipeline_properties or {}
+        self._pipeline_parameters = pipeline_parameters or []
         self._operations = {}
 
     @property
@@ -401,6 +404,13 @@ def pipeline_properties(self) -> Dict[str, Any]:
         """
         return self._pipeline_properties
 
+    @property
+    def pipeline_parameters(self) -> ElyraPropertyList[PipelineParameter]:
+        """
+        The list of parameters associated with this pipeline
+        """
+        return self._pipeline_parameters
+
     @property
     def operations(self) -> Dict[str, Operation]:
         return self._operations
@@ -432,46 +442,3 @@ def __eq__(self, other: "Pipeline") -> bool:
                 and self.runtime == other.runtime
                 and self.operations == other.operations
             )
-
-
-class KeyValueList(list):
-    """
-    A list class that exposes functionality specific to lists whose entries are
-    key-value pairs separated by a pre-defined character.
-    """
-
-    _key_value_separator: str = "="
-
-    def to_dict(self) -> Dict[str, str]:
-        """
-        Properties consisting of key-value pairs are stored in a list of separated
-        strings, while most processing steps require a dictionary - so we must convert.
-        If no key/value pairs are specified, an empty dictionary is returned, otherwise
-        pairs are converted to dictionary entries, stripped of whitespace, and returned.
-        """
-        kv_dict = {}
-        for kv in self:
-            if not kv:
-                continue
-
-            if self._key_value_separator not in kv:
-                raise ValueError(
-                    f"Property {kv} does not contain the expected "
-                    f"separator character: '{self._key_value_separator}'."
-                )
-
-            key, value = kv.split(self._key_value_separator, 1)
-
-            key = key.strip()
-            if not key:
-                # Invalid entry; skip inclusion and continue
-                continue
-
-            if isinstance(value, str):
-                value = value.strip()
-            if not value:
-                # Invalid entry; skip inclusion and continue
-                continue
-
-            kv_dict[key] = value
-        return kv_dict

elyra/pipeline/pipeline_constants.py

@@ -15,6 +15,7 @@
 #
 
 PIPELINE_DEFAULTS = "pipeline_defaults"
+PIPELINE_PARAMETERS = "pipeline_parameters"
 RUNTIME_IMAGE = "runtime_image"
 ENV_VARIABLES = "env_vars"
 MOUNTED_VOLUMES = "mounted_volumes"