feat: add secret parameter type for pipeline parameters(HEXA-1457 )

openhexa/graphql/graphql_client/enums.py

@@ -571,6 +571,7 @@ class ParameterType(str, Enum):
     int = "int"
     postgresql = "postgresql"
     s3 = "s3"
+    secret = "secret"
     str = "str"
 
 

openhexa/graphql/schema.generated.graphql

@@ -3180,6 +3180,7 @@ enum ParameterType {
   int
   postgresql
   s3
+  secret
   str
 }
 

openhexa/sdk/__init__.py

@@ -3,7 +3,7 @@
 from .datasets import Dataset
 from .files import File
 from .pipelines import current_pipeline, current_run, parameter, pipeline
-from .pipelines.parameter import DHIS2Widget, IASOWidget
+from .pipelines.parameter import DHIS2Widget, IASOWidget, Secret
 from .utils import OpenHexaClient
 from .workspaces import workspace
 from .workspaces.connection import (
@@ -32,4 +32,5 @@
     "Dataset",
     "OpenHexaClient",
     "File",
+    "Secret",
 ]

openhexa/sdk/pipelines/parameter.py

@@ -389,11 +389,74 @@ def validate(self, value: typing.Any | None) -> File:
             raise ParameterValueError(str(e))
 
 
+class Secret:
+    """Marker type for secret/password pipeline parameters.
+
+    Use as the ``type`` argument of the ``@parameter`` decorator to indicate that the parameter value is sensitive
+    and should be hidden in the OpenHEXA web interface. The pipeline function will receive the value as a plain
+    ``str`` at runtime.
+
+    Example::
+
+        @parameter("iaso_token", type=Secret, name="IASO token", required=True)
+        @pipeline("my-pipeline")
+        def my_pipeline(iaso_token: str):
+            ...
+    """
+
+    pass
+
+
+class SecretType(ParameterType):
+    """Type class for secret/password string parameters. Values are treated as plain strings at runtime."""
+
+    @property
+    def spec_type(self) -> str:
+        """Return a type string for the specs that are sent to the backend."""
+        return "secret"
+
+    @property
+    def expected_type(self) -> type:
+        """Returns the python type expected for values."""
+        return str
+
+    @property
+    def accepts_choices(self) -> bool:
+        """Secrets don't support choices."""
+        return False
+
+    @property
+    def accepts_multiple(self) -> bool:
+        """Secrets don't support multiple values."""
+        return False
+
+    @staticmethod
+    def normalize(value: typing.Any) -> str | None:
+        """Strip whitespace and convert empty strings to None."""
+        if isinstance(value, str):
+            normalized_value = value.strip()
+        else:
+            normalized_value = value
+
+        if normalized_value == "":
+            return None
+
+        return normalized_value
+
+    def validate_default(self, value: typing.Any | None):
+        """Validate the default value configured for this type."""
+        if value == "":
+            raise ParameterValueError("Empty values are not accepted.")
+
+        super().validate_default(value)
+
+
 TYPES_BY_PYTHON_TYPE = {
     "str": StringType,
     "bool": Boolean,
     "int": Integer,
     "float": Float,
+    "Secret": SecretType,
     "DHIS2Connection": DHIS2ConnectionType,
     "PostgreSQLConnection": PostgreSQLConnectionType,
     "IASOConnection": IASOConnectionType,
@@ -438,6 +501,7 @@ def __init__(
             | int
             | bool
             | float
+            | Secret
             | DHIS2Connection
             | IASOConnection
             | PostgreSQLConnection
@@ -460,7 +524,10 @@ def __init__(
         self.code = code
 
         try:
-            self.type = TYPES_BY_PYTHON_TYPE[type.__name__]()
+            if isinstance(type, ParameterType):
+                self.type = type
+            else:
+                self.type = TYPES_BY_PYTHON_TYPE[type.__name__]()
         except (KeyError, AttributeError):
             valid_parameter_types = [k for k in TYPES_BY_PYTHON_TYPE.keys()]
             raise InvalidParameterError(
@@ -621,6 +688,7 @@ def parameter(
         | int
         | bool
         | float
+        | Secret
         | DHIS2Connection
         | IASOConnection
         | PostgreSQLConnection

openhexa/sdk/pipelines/runtime.py

@@ -308,7 +308,7 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
                 # Convert args spec to parameter kwargs
                 param_kwargs = {k: v["value"] for k, v in parameter_args.items()}
 
-                parameter = Parameter(type=type_class.expected_type, **param_kwargs)
+                parameter = Parameter(type=type_class, **param_kwargs)
                 pipeline_parameters.append(parameter)
 
             except KeyError as e:

tests/test_parameter.py

@@ -32,6 +32,8 @@
     ParameterValueError,
     PostgreSQLConnectionType,
     S3ConnectionType,
+    Secret,
+    SecretType,
     StringType,
     parameter,
 )
@@ -92,6 +94,54 @@ def test_parameter_types_validate():
         boolean_parameter_type.validate(86)
 
 
+def test_secret_type_normalize():
+    """Check normalization for SecretType."""
+    secret_type = SecretType()
+    assert secret_type.normalize("my-token") == "my-token"
+    assert secret_type.normalize("  my-token  ") == "my-token"
+    assert secret_type.normalize("") is None
+    assert secret_type.normalize(" ") is None
+
+
+def test_secret_type_validate():
+    """Check validation for SecretType."""
+    secret_type = SecretType()
+    assert secret_type.validate("my-token") == "my-token"
+    with pytest.raises(ParameterValueError):
+        secret_type.validate(123)
+
+
+def test_secret_type_does_not_accept_choices():
+    """Secret parameters don't support choices."""
+    with pytest.raises(InvalidParameterError):
+        Parameter("token", type=Secret, choices=["a", "b"])
+
+
+def test_secret_type_does_not_accept_multiple():
+    """Secret parameters don't support multiple values."""
+    with pytest.raises(InvalidParameterError):
+        Parameter("token", type=Secret, multiple=True)
+
+
+def test_secret_parameter_spec_type():
+    """Secret parameters serialize with spec_type 'secret'."""
+    p = Parameter("token", type=Secret)
+    assert p.to_dict()["type"] == "secret"
+
+
+def test_secret_parameter_validates_string():
+    """Secret parameters validate and return plain strings."""
+    p = Parameter("token", type=Secret)
+    assert p.validate("my-secret-token") == "my-secret-token"
+
+
+def test_secret_parameter_required():
+    """Secret parameters respect required constraint."""
+    p = Parameter("token", type=Secret, required=True)
+    with pytest.raises(ParameterValueError):
+        p.validate(None)
+
+
 def test_validate_postgres_connection():
     """Check PostgreSQL connection validation."""
     identifier = "polio-ff3a0d"