feature: 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/choices.py

@@ -0,0 +1,55 @@
+"""Dynamic choices classes for pipeline parameters."""
+
+from openhexa.sdk.pipelines.exceptions import InvalidParameterError
+
+_SUPPORTED_FORMATS = {"csv", "json", "yaml", "yml"}
+
+
+class ChoicesFromFile:
+    """Descriptor for choices loaded dynamically from a file in the workspace file system.
+
+    The file format is inferred from the path extension (.csv, .json, .yaml, .yml).
+    For CSV files with a single column, that column is used automatically.
+    For CSV/JSON/YAML files with multiple columns/keys, `column` must be specified.
+
+    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.
+    """
+
+    def __init__(self, path: str, column: str | None = None):
+        self.path = path
+        self.column = column
+        self.format = self._detect_format(path)
+        self.validate_spec()
+
+    @staticmethod
+    def _detect_format(path: str) -> str:
+        if not path or not isinstance(path, str):
+            raise InvalidParameterError("ChoicesFromFile path must be a non-empty string.")
+        ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
+        if ext not in _SUPPORTED_FORMATS:
+            raise InvalidParameterError(
+                f"Cannot determine file format from path '{path}'. "
+                f"Supported extensions: {', '.join(sorted(_SUPPORTED_FORMATS))}."
+            )
+        return "yaml" if ext == "yml" else ext
+
+    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.")
+
+    def to_dict(self) -> dict:
+        """Return a dictionary representation of the choices spec."""
+        return {
+            "format": self.format,
+            "path": self.path,
+            "column": self.column,
+        }