feat: SDK - pipeline run parameters - dynamic choices (HEXA-1620)

.gitignore

@@ -2,6 +2,7 @@
 __pycache__/
 *.py[cod]
 *$py.class
+.idea/
 
 # C extensions
 *.so

Makefile

@@ -0,0 +1,3 @@
+l lint:
+	@echo "Executing lint in backend code (pre-commit)"
+	pre-commit run --show-diff-on-failure --color=always --all-files

openhexa/sdk/pipelines/parameter/__init__.py

@@ -0,0 +1,68 @@
+"""Pipeline parameters classes and functions.
+
+See https://github.com/BLSQ/openhexa/wiki/Writing-OpenHEXA-pipelines#pipeline-parameters for more information.
+"""
+
+from openhexa.sdk.pipelines.exceptions import InvalidParameterError, ParameterValueError
+
+from .choices import ChoicesFromFile
+from .decorator import FunctionWithParameter, Parameter, parameter, validate_parameters
+from .types import (
+    TYPES_BY_PYTHON_TYPE,
+    Boolean,
+    ConnectionParameterType,
+    CustomConnectionType,
+    DatasetType,
+    DHIS2ConnectionType,
+    FileType,
+    Float,
+    GCSConnectionType,
+    IASOConnectionType,
+    Integer,
+    ParameterType,
+    PostgreSQLConnectionType,
+    S3ConnectionType,
+    Secret,
+    SecretType,
+    StringType,
+)
+from .widgets import DHIS2Widget, IASOWidget
+
+__all__ = [
+    # Decorator and core classes
+    "parameter",
+    "Parameter",
+    "FunctionWithParameter",
+    "validate_parameters",
+    # Type base classes
+    "ParameterType",
+    "ConnectionParameterType",
+    # Primitive types
+    "StringType",
+    "Boolean",
+    "Integer",
+    "Float",
+    # Connection types
+    "PostgreSQLConnectionType",
+    "S3ConnectionType",
+    "GCSConnectionType",
+    "DHIS2ConnectionType",
+    "IASOConnectionType",
+    "CustomConnectionType",
+    # Resource types
+    "DatasetType",
+    "FileType",
+    # Secret
+    "Secret",
+    "SecretType",
+    # Registry
+    "TYPES_BY_PYTHON_TYPE",
+    # Dynamic choices
+    "ChoicesFromFile",
+    # Widgets
+    "DHIS2Widget",
+    "IASOWidget",
+    # Exceptions (re-exported for backward compat)
+    "InvalidParameterError",
+    "ParameterValueError",
+]

openhexa/sdk/pipelines/parameter/ast_constructible.py

@@ -0,0 +1,44 @@
+"""Mixin for classes that can reconstruct themselves from an AST Call node."""
+
+import ast
+import inspect
+
+
+class AstConstructible:
+    """Mixin that enables reconstruction of a class instance from an AST Call node.
+
+    Any class whose ``__init__`` takes only scalar (``ast.Constant``) arguments
+    can inherit from this mixin and get ``from_ast_call`` for free.  Adding or
+    renaming ``__init__`` parameters does *not* require touching the parser.
+
+    To make the AST parser recognise a new subclass by name, add one entry to
+    ``_AST_CALLABLE_TYPES`` in ``runtime.py`` (and ensure the subclass module is
+    imported there).  Auto-registration via ``__init_subclass__`` would not remove
+    that requirement — the registry entry only exists after the module is imported,
+    so an explicit import would still be needed.
+    """
+
+    @classmethod
+    def from_ast_call(cls, node: ast.Call) -> "AstConstructible":
+        """Reconstruct an instance from an AST Call node.
+
+        Maps positional args to ``__init__`` parameter names via
+        ``inspect.signature``, then merges keyword args, and calls ``cls``.
+        """
+        param_names = list(inspect.signature(cls).parameters.keys())
+        kwargs = {}
+        for i, arg in enumerate(node.args):
+            if i >= len(param_names):
+                break
+            if not isinstance(arg, ast.Constant):
+                raise ValueError(
+                    f"{cls.__name__}() positional argument {i + 1} must be a literal value, not a dynamic expression."
+                )
+            kwargs[param_names[i]] = arg.value
+        for kw in node.keywords:
+            if not isinstance(kw.value, ast.Constant):
+                raise ValueError(
+                    f"{cls.__name__}() keyword argument '{kw.arg}' must be a literal value, not a dynamic expression."
+                )
+            kwargs[kw.arg] = kw.value.value
+        return cls(**kwargs)

openhexa/sdk/pipelines/parameter/choices.py

@@ -0,0 +1,67 @@
+"""Dynamic choices classes for pipeline parameters."""
+
+from openhexa.sdk.pipelines.exceptions import InvalidParameterError
+
+from .ast_constructible import AstConstructible
+
+_SUPPORTED_FORMATS = {"csv", "json", "yaml", "yml"}
+
+
+class ChoicesFromFile(AstConstructible):
+    """Descriptor for choices loaded dynamically from a file in the workspace file system.
+
+    Parameters
+    ----------
+    path : str
+        Path to the file in the workspace file system (e.g. "data/districts.csv").
+    column : str, optional
+        Column name (CSV) or key (JSON/YAML) to use as choice values.
+        Required when the file has more than one column/key.
+    format : str, optional
+        File format (e.g. "csv", "json", "yaml"). Sent as-is to the platform.
+    """
+
+    def __init__(self, path: str, column: str | None = None, format: str | None = None):
+        self.path = path
+        self.column = column
+        self.format = format
+        self._validate_spec()
+
+    def _validate_spec(self):
+        """Validate the path and column specification."""
+        if not self.path or not isinstance(self.path, str):
+            raise InvalidParameterError("ChoicesFromFile path must be a non-empty string.")
+        if self.column is not None and not isinstance(self.column, str):
+            raise InvalidParameterError("ChoicesFromFile column must be a string.")
+        if self.format is not None and self.format not in _SUPPORTED_FORMATS:
+            raise InvalidParameterError(
+                f"ChoicesFromFile format '{self.format}' is not supported. "
+                f"Supported formats: {', '.join(sorted(_SUPPORTED_FORMATS))}."
+            )
+
+    def __repr__(self) -> str:
+        """Return a string representation of the ChoicesFromFile instance."""
+        parts = [repr(self.path)]
+        if self.column is not None:
+            parts.append(f"column={self.column!r}")
+        if self.format is not None:
+            parts.append(f"format={self.format!r}")
+        return f"ChoicesFromFile({', '.join(parts)})"
+
+    def __eq__(self, other: object) -> bool:
+        """Check equality based on path, column, and format."""
+        if not isinstance(other, ChoicesFromFile):
+            return NotImplemented
+        return self.path == other.path and self.column == other.column and self.format == other.format
+
+    def __hash__(self) -> int:
+        """Return hash based on path, column, and format."""
+        return hash((self.path, self.column, self.format))
+
+    def to_dict(self) -> dict:
+        """Return a dictionary representation of the choices spec."""
+        return {
+            "format": self.format,
+            "path": self.path,
+            "column": self.column,
+        }