feat(Parameter): HEXA-1134 adds dhis2 parameters widget and connection fields plus validation (#225)

* remove deprecated parameter specs

* add connection field validation

* apply linting

* apply linting

* fix: assert test for widget parameter

* only check supported conneciton types

* use dict and change error message

Author: Nazar F

openhexa/cli/api.py

@@ -596,3 +596,23 @@ def upload_pipeline(
             raise Exception(data["uploadPipeline"]["errors"])
 
     return data["uploadPipeline"]["pipelineVersion"]
+
+
+def is_dhis2_connection_up(workspace_slug: str, connection_slug: str) -> bool:
+    """DHIS2 connection status."""
+    response = graphql(
+        """
+        query getConnectionBySlug($workspaceSlug: String!, $connectionSlug: String!) {
+        connectionBySlug(workspaceSlug:$workspaceSlug, connectionSlug: $connectionSlug){
+            ... on DHIS2Connection {
+                    status
+                }
+        }
+        }
+        """,
+        variables={
+            "workspaceSlug": workspace_slug,
+            "connectionSlug": connection_slug,
+        },
+    )
+    response["data"]["connectionBySlug"]["status"] == "UP"

openhexa/sdk/pipelines/log_level.py

@@ -1,4 +1,5 @@
 """Log levels for the pipeline runs."""
+
 from enum import IntEnum
 
 

openhexa/sdk/pipelines/parameter.py

@@ -58,7 +58,6 @@ def validate(self, value: typing.Optional[typing.Any]) -> typing.Optional[typing
             raise ParameterValueError(
                 f"Invalid type for value {value} (expected {self.expected_type}, got {type(value)})"
             )
-
         return value
 
     def validate_default(self, value: typing.Optional[typing.Any]):
@@ -382,6 +381,8 @@ def __init__(
         choices: typing.Optional[typing.Sequence] = None,
         help: typing.Optional[str] = None,
         default: typing.Optional[typing.Any] = None,
+        widget: typing.Optional[str] = None,
+        connection: typing.Optional[str] = None,
         required: bool = True,
         multiple: bool = False,
     ):
@@ -419,6 +420,9 @@ def __init__(
             raise InvalidParameterError(f"Parameters of type {self.type} can't have multiple values.")
         self.multiple = multiple
 
+        self.widget = widget
+        self.connection = connection
+
         self._validate_default(default, multiple)
         self.default = default
 
@@ -438,6 +442,8 @@ def to_dict(self) -> dict[str, typing.Any]:
             "choices": self.choices,
             "help": self.help,
             "default": self.default,
+            "widget": self.widget,
+            "connection": self.connection,
             "required": self.required,
             "multiple": self.multiple,
         }
@@ -511,18 +517,17 @@ def _validate_default(self, default: typing.Any, multiple: bool):
                     f"The default value for {self.code} is not included in the provided choices."
                 )
 
-    def parameter_spec(self) -> dict[str, typing.Any]:
-        """Build specification for this parameter, to be provided to the OpenHEXA backend."""
-        return {
-            "type": self.type.spec_type,
-            "required": self.required,
-            "choices": self.choices,
-            "code": self.code,
-            "name": self.name,
-            "help": self.help,
-            "multiple": self.multiple,
-            "default": self.default,
-        }
+
+def validate_parameters_with_connection(parameters: [Parameter]):
+    """Validate the provided connection parameters if they relate to existing connection parameter."""
+    supported_connection_types = {DHIS2ConnectionType}
+    connection_parameters = {p.code for p in parameters if type(p.type) in supported_connection_types}
+
+    for parameter in parameters:
+        if parameter.connection and parameter.connection not in connection_parameters:
+            raise InvalidParameterError(
+                f"Connection field '{parameter.code}' references a non-existing connection parameter '{parameter.connection}'"
+            )
 
 
 def parameter(

openhexa/sdk/pipelines/pipeline.py

@@ -163,10 +163,6 @@ def run(self, config: dict[str, typing.Any]):
         now = datetime.datetime.now(tz=datetime.timezone.utc).replace(microsecond=0).isoformat()
         print(f'{now} Successfully completed pipeline "{self.code}"')
 
-    def parameters_spec(self) -> list[dict[str, typing.Any]]:
-        """Return the individual specifications of all the parameters of this pipeline."""
-        return [arg.parameter_spec() for arg in self.parameters]
-
     def to_dict(self):
         """Return a dictionary representation of the pipeline."""
         return {

openhexa/sdk/pipelines/runtime.py

@@ -15,7 +15,7 @@
 import requests
 
 from openhexa.sdk.pipelines.exceptions import PipelineNotFound
-from openhexa.sdk.pipelines.parameter import TYPES_BY_PYTHON_TYPE, Parameter
+from openhexa.sdk.pipelines.parameter import TYPES_BY_PYTHON_TYPE, Parameter, validate_parameters_with_connection
 
 from .pipeline import Pipeline
 
@@ -147,6 +147,8 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
                         Argument("choices", [ast.List]),
                         Argument("help", [ast.Constant]),
                         Argument("default", [ast.Constant, ast.List]),
+                        Argument("widget", [ast.Constant]),
+                        Argument("connection", [ast.Constant]),
                         Argument("required", [ast.Constant], default_value=True),
                         Argument("multiple", [ast.Constant], default_value=False),
                     ),
@@ -159,6 +161,8 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
                 parameter = Parameter(type=type_class.expected_type, **parameter_args)
                 pipelines_parameters.append(parameter)
 
+            validate_parameters_with_connection(pipelines_parameters)
+
             pipeline = Pipeline(parameters=pipelines_parameters, function=None, **pipeline_decorator_spec["args"])
 
     if pipeline is None:

tests/test_ast.py

@@ -3,7 +3,7 @@
 import tempfile
 from unittest import TestCase
 
-from openhexa.sdk.pipelines.exceptions import PipelineNotFound
+from openhexa.sdk.pipelines.exceptions import InvalidParameterError, PipelineNotFound
 from openhexa.sdk.pipelines.runtime import get_pipeline
 
 
@@ -147,6 +147,8 @@ def test_pipeline_with_int_param(self):
                             "type": "int",
                             "name": "Test Param",
                             "default": 42,
+                            "widget": None,
+                            "connection": None,
                             "help": "Param help",
                             "required": True,
                         }
@@ -188,6 +190,8 @@ def test_pipeline_with_multiple_param(self):
                             "type": "int",
                             "name": "Test Param",
                             "default": [42],
+                            "widget": None,
+                            "connection": None,
                             "help": "Param help",
                             "required": True,
                         }
@@ -230,6 +234,8 @@ def test_pipeline_with_dataset(self):
                             "type": "dataset",
                             "name": "Dataset",
                             "default": None,
+                            "widget": None,
+                            "connection": None,
                             "help": "Dataset",
                             "required": False,
                         }
@@ -271,6 +277,8 @@ def test_pipeline_with_choices(self):
                             "type": "str",
                             "name": "Test Param",
                             "default": None,
+                            "widget": None,
+                            "connection": None,
                             "help": "Param help",
                             "required": True,
                         }
@@ -340,6 +348,8 @@ def test_pipeline_with_bool(self):
                             "type": "bool",
                             "name": "Test Param",
                             "default": True,
+                            "widget": None,
+                            "connection": None,
                             "help": "Param help",
                             "required": True,
                         }
@@ -382,6 +392,8 @@ def test_pipeline_with_multiple_parameters(self):
                             "type": "int",
                             "name": "Test Param",
                             "default": 42,
+                            "widget": None,
+                            "connection": None,
                             "help": "Param help",
                             "required": True,
                         },
@@ -392,6 +404,8 @@ def test_pipeline_with_multiple_parameters(self):
                             "type": "str",
                             "name": "Test Param 2",
                             "default": None,
+                            "widget": None,
+                            "connection": None,
                             "help": "Param help 2",
                             "required": True,
                         },
@@ -419,3 +433,124 @@ def test_pipeline_with_unsupported_parameter(self):
                 )
             with self.assertRaises(KeyError):
                 get_pipeline(tmpdirname)
+
+    def test_pipeline_with_connection_parameter(self):
+        """The file contains a @pipeline decorator and a @parameter decorator with a connection type."""
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            with open(f"{tmpdirname}/pipeline.py", "w") as f:
+                f.write(
+                    "\n".join(
+                        [
+                            "from openhexa.sdk.pipelines import pipeline, parameter",
+                            "",
+                            "@parameter('dhis_con', name='DHIS2 Connection', type=DHIS2Connection, required=True)",
+                            "@parameter('data_element_ids', name='Data Elements id', type=str, widget='dhis2.data_elements.picker', connection='dhis_con', required=True)",
+                            "@pipeline('test', 'Test pipeline')",
+                            "def test_pipeline():",
+                            "    pass",
+                            "",
+                        ]
+                    )
+                )
+            pipeline = get_pipeline(tmpdirname)
+            self.maxDiff = None
+            self.assertEqual(
+                pipeline.to_dict(),
+                {
+                    "code": "test",
+                    "name": "Test pipeline",
+                    "function": None,
+                    "tasks": [],
+                    "parameters": [
+                        {
+                            "code": "dhis_con",
+                            "type": "dhis2",
+                            "name": "DHIS2 Connection",
+                            "default": None,
+                            "multiple": False,
+                            "choices": None,
+                            "widget": None,
+                            "connection": None,
+                            "help": None,
+                            "required": True,
+                        },
+                        {
+                            "code": "data_element_ids",
+                            "type": "str",
+                            "name": "Data Elements id",
+                            "widget": "dhis2.data_elements.picker",
+                            "connection": "dhis_con",
+                            "default": None,
+                            "multiple": False,
+                            "choices": None,
+                            "help": None,
+                            "required": True,
+                        },
+                    ],
+                    "timeout": None,
+                },
+            )
+
+    def test_pipeline_wit_wrong_connection_parameter(self):
+        """The file contains a @pipeline decorator and a @parameter decorator with a non-existing connection type."""
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            with open(f"{tmpdirname}/pipeline.py", "w") as f:
+                f.write(
+                    "\n".join(
+                        [
+                            "from openhexa.sdk.pipelines import pipeline, parameter",
+                            "",
+                            "@parameter('dhis_con', name='DHIS2 Connection', type=DHIS2Connection, required=True)",
+                            "@parameter('data_element_ids', name='Data Elements id', type=str, widget='dhis2.data_elements.picker', connection='sds_con', required=True)",
+                            "@pipeline('test', 'Test pipeline')",
+                            "def test_pipeline():",
+                            "    pass",
+                            "",
+                        ]
+                    )
+                )
+            with self.assertRaises(InvalidParameterError):
+                get_pipeline(tmpdirname)
+
+    def test_pipeline_with_widget_without_connection(self):
+        """The file contains a @pipeline decorator and a @parameter decorator with a widget parameter field."""
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            with open(f"{tmpdirname}/pipeline.py", "w") as f:
+                f.write(
+                    "\n".join(
+                        [
+                            "from openhexa.sdk.pipelines import pipeline, parameter",
+                            "",
+                            "@parameter('test_field_for_wdiget', name='Widget Param', type=str, widget='custom_picker', help='Param help')",
+                            "@pipeline('test', 'Test pipeline')",
+                            "def test_pipeline():",
+                            "    pass",
+                            "",
+                        ]
+                    )
+                )
+            pipeline = get_pipeline(tmpdirname)
+            self.assertEqual(
+                pipeline.to_dict(),
+                {
+                    "code": "test",
+                    "name": "Test pipeline",
+                    "function": None,
+                    "tasks": [],
+                    "parameters": [
+                        {
+                            "code": "test_field_for_wdiget",
+                            "type": "str",
+                            "name": "Widget Param",
+                            "default": None,
+                            "multiple": False,
+                            "choices": None,
+                            "widget": "custom_picker",
+                            "connection": None,
+                            "help": "Param help",
+                            "required": True,
+                        }
+                    ],
+                    "timeout": None,
+                },
+            )

tests/test_parameter.py

@@ -316,43 +316,6 @@ def test_parameter_validate_multiple():
         parameter_4.validate(["ab", "xy"])
 
 
-def test_parameter_parameters_spec():
-    """Verify that parameter specifications are built properly and have the proper defaults."""
-    # required is True by default
-    an_parameter = Parameter("arg1", type=str, default="yep")
-    another_parameter = Parameter(
-        "arg2",
-        type=str,
-        name="Arg 2",
-        help="Help 2",
-        choices=["ab", "cd"],
-        required=False,
-        multiple=True,
-    )
-
-    assert an_parameter.parameter_spec() == {
-        "code": "arg1",
-        "name": None,
-        "type": "str",
-        "required": True,
-        "choices": None,
-        "help": None,
-        "multiple": False,
-        "default": "yep",
-    }
-
-    assert another_parameter.parameter_spec() == {
-        "code": "arg2",
-        "name": "Arg 2",
-        "type": "str",
-        "required": False,
-        "choices": ["ab", "cd"],
-        "help": "Help 2",
-        "multiple": True,
-        "default": None,
-    }
-
-
 def test_parameter_decorator():
     """Ensure that the @parameter decorator behaves as expected (options and defaults)."""