chore: Better Validation of the DHIS2 widgets (#245)

* chore(dhis2): add an example pipeline for dhis2 widgets

* refactor(Parameter): rename validate_parameters_with_connection to validate_parameters and enhance validation for DHIS2 widgets requiring a connection parameter

* Update pipeline.py docstring to clarify purpose as an example for retrieving data elements from DHIS2.

* refactor: replace ParameterWidget with DHIS2Widget for improved clarity and organization in DHIS2 pipeline parameters

* refactor: remove comment

Author: qgerome

examples/pipelines/dhis2/.gitignore

@@ -0,0 +1,2 @@
+workspace/
+workspace.yaml

examples/pipelines/dhis2/pipeline.py

@@ -0,0 +1,32 @@
+"""Example pipeline to get data elements from DHIS2."""
+
+from openhexa.sdk import current_run, parameter, pipeline
+from openhexa.sdk.pipelines.parameter import DHIS2Widget
+from openhexa.sdk.workspaces.connection import DHIS2Connection
+
+
+@pipeline("dhis2")
+@parameter("dhis2_con", type=DHIS2Connection, required=True)
+@parameter(
+    "data_elements",
+    type=str,
+    required=True,
+    widget=DHIS2Widget.DATA_ELEMENTS,
+    multiple=True,
+    connection="dhis2_con",
+)
+def dhis2(dhis2_con, data_elements):
+    """Get data elements from DHIS2."""
+    print_data_elements(dhis2_con, data_elements)
+
+
+@dhis2.task
+def print_data_elements(dhis2_con, data_elements):
+    """Print data elements."""
+    current_run.log_info("Printing data elements")
+
+    current_run.log_info(f"Data elements: {data_elements}")
+
+
+if __name__ == "__main__":
+    dhis2()

openhexa/sdk/pipelines/parameter.py

@@ -359,23 +359,17 @@ def validate(self, value: typing.Any | None) -> Dataset:
 }
 
 
-class ParameterWidget(StrEnum):
-    """
-    Enum for parameter available parameter widgets.
-
-    The list of supported widgets can be found in the OpenHEXA documentation or GraphQL schema.
-    /graphql/ -> __schema -> types -> ParameterWidget
-    https://github.com/blsq/openhexa/wiki/Writing-OpenHEXA-pipelines#using-widget-parameters
-    """
+class DHIS2Widget(StrEnum):
+    """Enum for DHIS2 widgets."""
 
-    DHIS2_ORG_UNITS = "DHIS2_ORG_UNITS"
-    DHIS2_ORG_UNIT_GROUPS = "DHIS2_ORG_UNIT_GROUPS"
-    DHIS2_ORG_UNIT_LEVELS = "DHIS2_ORG_UNIT_LEVELS"
-    DHIS2_DATASETS = "DHIS2_DATASETS"
-    DHIS2_DATA_ELEMENTS = "DHIS2_DATA_ELEMENTS"
-    DHIS2_DATA_ELEMENT_GROUPS = "DHIS2_DATA_ELEMENT_GROUPS"
-    DHIS2_INDICATORS = "DHIS2_INDICATORS"
-    DHIS2_INDICATOR_GROUPS = "DHIS2_INDICATOR_GROUPS"
+    ORG_UNITS = "DHIS2_ORG_UNITS"
+    ORG_UNIT_GROUPS = "DHIS2_ORG_UNIT_GROUPS"
+    ORG_UNIT_LEVELS = "DHIS2_ORG_UNIT_LEVELS"
+    DATASETS = "DHIS2_DATASETS"
+    DATA_ELEMENTS = "DHIS2_DATA_ELEMENTS"
+    DATA_ELEMENT_GROUPS = "DHIS2_DATA_ELEMENT_GROUPS"
+    INDICATORS = "DHIS2_INDICATORS"
+    INDICATOR_GROUPS = "DHIS2_INDICATOR_GROUPS"
 
 
 class Parameter:
@@ -399,13 +393,12 @@ def __init__(
         choices: typing.Sequence | None = None,
         help: str | None = None,
         default: typing.Any | None = None,
-        widget: ParameterWidget | None = None,
+        widget: DHIS2Widget | None = None,
         connection: str | None = None,
         required: bool = True,
         multiple: bool = False,
     ):
         validate_pipeline_parameter_code(code)
-
         self.code = code
 
         try:
@@ -460,7 +453,7 @@ def to_dict(self) -> dict[str, typing.Any]:
             "choices": self.choices,
             "help": self.help,
             "default": self.default,
-            "widget": self.widget.value if self.widget else None,
+            "widget": self.widget if self.widget else None,
             "connection": self.connection,
             "required": self.required,
             "multiple": self.multiple,
@@ -536,7 +529,7 @@ def _validate_default(self, default: typing.Any, multiple: bool):
                 )
 
 
-def validate_parameters_with_connection(parameters: [Parameter]):
+def validate_parameters(parameters: list[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}
@@ -546,6 +539,11 @@ def validate_parameters_with_connection(parameters: [Parameter]):
             raise InvalidParameterError(
                 f"Connection field '{parameter.code}' references a non-existing connection parameter '{parameter.connection}'"
             )
+        if parameter.widget and parameter.widget in DHIS2Widget and not parameter.connection:
+            raise InvalidParameterError(
+                f"DHIS2 widgets require a connection parameter. Please provide a connection parameter for {parameter.code}. "
+                f"Example: @parameter('{parameter.code}', type=str, widget=DHIS2Widget.{parameter.widget}, connection='my_connection')"
+            )
 
 
 def parameter(
@@ -565,7 +563,7 @@ def parameter(
     name: str | None = None,
     choices: typing.Sequence | None = None,
     help: str | None = None,
-    widget: ParameterWidget | None = None,
+    widget: DHIS2Widget | None = None,
     default: typing.Any | None = None,
     required: bool = True,
     multiple: bool = False,

openhexa/sdk/pipelines/runtime.py

@@ -16,9 +16,9 @@
 from openhexa.sdk.pipelines.exceptions import InvalidParameterError, PipelineNotFound
 from openhexa.sdk.pipelines.parameter import (
     TYPES_BY_PYTHON_TYPE,
+    DHIS2Widget,
     Parameter,
-    ParameterWidget,
-    validate_parameters_with_connection,
+    validate_parameters,
 )
 
 from .pipeline import Pipeline
@@ -168,7 +168,10 @@ def _get_decorator_arg_value(decorator: ast.Call, arg: Argument, index: int) ->
             elif isinstance(keyword.value, ast.List):
                 return ([el.value for el in keyword.value.elts], True)
             elif isinstance(keyword.value, ast.Attribute):
-                return (ParameterWidget(keyword.value.attr), True)
+                if keyword.value.attr in DHIS2Widget.__members__:
+                    return getattr(DHIS2Widget, keyword.value.attr), True
+                else:
+                    raise ValueError(f"Unsupported widget: {keyword.value.attr}")
 
     # Then check for positional arguments
     try:
@@ -303,8 +306,7 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
             except KeyError as e:
                 raise InvalidParameterError(f"Missing required parameter attribute: {e}")
 
-        # Validate parameters with connections
-        validate_parameters_with_connection(pipeline_parameters)
+        validate_parameters(pipeline_parameters)
 
         # Create and return the pipeline
         return Pipeline(

tests/test_ast.py

@@ -6,7 +6,7 @@
 from unittest.mock import patch
 
 from openhexa.sdk.pipelines.exceptions import InvalidParameterError, PipelineNotFound
-from openhexa.sdk.pipelines.parameter import ParameterWidget
+from openhexa.sdk.pipelines.parameter import DHIS2Widget
 from openhexa.sdk.pipelines.runtime import get_pipeline
 
 
@@ -436,10 +436,10 @@ def test_pipeline_with_connection_parameter(self):
                     "\n".join(
                         [
                             "from openhexa.sdk.pipelines import pipeline, parameter",
-                            "from openhexa.sdk.pipelines.widgets import ParameterWidget",
+                            "from openhexa.sdk.pipelines.widgets import DHIS2Widget",
                             "",
                             "@parameter('dhis_con', name='DHIS2 Connection', type=DHIS2Connection, required=True)",
-                            "@parameter('data_element_ids', name='Data Elements id', type=str, widget=ParameterWidget.DHIS2_ORG_UNITS, connection='dhis_con', required=True)",
+                            "@parameter('data_element_ids', name='Data Elements id', type=str, widget=DHIS2Widget.ORG_UNITS, connection='dhis_con', required=True)",
                             "@pipeline('Test pipeline')",
                             "def test_pipeline():",
                             "    pass",
@@ -472,7 +472,7 @@ def test_pipeline_with_connection_parameter(self):
                             "code": "data_element_ids",
                             "type": "str",
                             "name": "Data Elements id",
-                            "widget": ParameterWidget.DHIS2_ORG_UNITS.value,
+                            "widget": DHIS2Widget.ORG_UNITS.value,
                             "connection": "dhis_con",
                             "default": None,
                             "multiple": False,
@@ -493,11 +493,11 @@ def test_pipeline_wit_wrong_connection_parameter(self):
                     "\n".join(
                         [
                             "from openhexa.sdk.pipelines import pipeline, parameter",
-                            "from openhexa.sdk.pipelines.parameter import ParameterWidget",
+                            "from openhexa.sdk.pipelines.parameter import DHIS2Widget",
                             "",
                             "@parameter('dhis_con', name='DHIS2 Connection', type=DHIS2Connection, required=True)",
                             "@pipeline('Test pipeline')",
-                            "@parameter('data_element_ids', name='Data Elements id', type=str, widget=ParameterWidget.DHIS2_ORG_UNITS, connection='sds_con', required=True)",
+                            "@parameter('data_element_ids', name='Data Elements id', type=str, widget=DHIS2Widget.ORG_UNITS, connection='sds_con', required=True)",
                             "def test_pipeline():",
                             "    pass",
                             "",
@@ -507,48 +507,26 @@ def test_pipeline_wit_wrong_connection_parameter(self):
             with self.assertRaises(InvalidParameterError):
                 get_pipeline(tmpdirname)
 
-    def test_pipeline_with_widget_without_connection(self):
+    def test_pipeline_with_dhis2_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",
-                            "from openhexa.sdk.pipelines.parameter import ParameterWidget",
+                            "from openhexa.sdk.pipelines.parameter import DHIS2Widget",
                             "",
-                            "@parameter('test_field_for_widget', name='Widget Param', type=str, widget=ParameterWidget.DHIS2_ORG_UNITS, help='Param help')",
+                            "@parameter('test_field_for_widget', name='Widget Param', type=str, widget=DHIS2Widget.ORG_UNITS, help='Param help')",
                             "@pipeline('Test pipeline')",
                             "def test_pipeline():",
                             "    pass",
                             "",
                         ]
                     )
                 )
-            pipeline = get_pipeline(tmpdirname)
-            self.assertEqual(
-                pipeline.to_dict(),
-                {
-                    "name": "Test pipeline",
-                    "function": None,
-                    "tasks": [],
-                    "parameters": [
-                        {
-                            "code": "test_field_for_widget",
-                            "type": "str",
-                            "name": "Widget Param",
-                            "default": None,
-                            "multiple": False,
-                            "choices": None,
-                            "widget": ParameterWidget.DHIS2_ORG_UNITS.value,
-                            "connection": None,
-                            "help": "Param help",
-                            "required": True,
-                        }
-                    ],
-                    "timeout": None,
-                },
-            )
+            with self.assertRaises(InvalidParameterError):
+                get_pipeline(tmpdirname)
 
     def test_pipeline_with_deprecated_code_argument_with_name(self):
         """The file contains a @pipeline decorator with the deprecated 'code' argument."""