feature: pipeline run parameters - dynamic choices (HEXA-1620)

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

@@ -5,6 +5,7 @@
 
 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,
@@ -56,6 +57,8 @@
     "SecretType",
     # Registry
     "TYPES_BY_PYTHON_TYPE",
+    # Dynamic choices
+    "ChoicesFromFile",
     # Widgets
     "DHIS2Widget",
     "IASOWidget",

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,71 @@
+"""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.
+
+    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._validate_spec()
+        self.format = self._detect_format(path)
+
+    @staticmethod
+    def _detect_format(path: str) -> str:
+        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 __repr__(self) -> str:
+        """Return a string representation of the ChoicesFromFile instance."""
+        if self.column is not None:
+            return f"ChoicesFromFile({self.path!r}, column={self.column!r})"
+        return f"ChoicesFromFile({self.path!r})"
+
+    def __eq__(self, other: object) -> bool:
+        """Check equality based on path and column."""
+        if not isinstance(other, ChoicesFromFile):
+            return NotImplemented
+        return self.path == other.path and self.column == other.column
+
+    def __hash__(self) -> int:
+        """Return hash based on path and column."""
+        return hash((self.path, self.column))
+
+    def to_dict(self) -> dict:
+        """Return a dictionary representation of the choices spec."""
+        return {
+            "format": self.format,
+            "path": self.path,
+            "column": self.column,
+        }

openhexa/sdk/pipelines/parameter/decorator.py

@@ -15,6 +15,7 @@
     S3Connection,
 )
 
+from .choices import ChoicesFromFile
 from .types import TYPES_BY_PYTHON_TYPE, Boolean, DHIS2ConnectionType, IASOConnectionType, Secret
 from .widgets import DHIS2Widget, IASOWidget
 
@@ -42,7 +43,7 @@ def __init__(
             | File
         ],
         name: str | None = None,
-        choices: typing.Sequence | None = None,
+        choices: typing.Sequence | ChoicesFromFile | str | None = None,
         help: str | None = None,
         default: typing.Any | None = None,
         widget: DHIS2Widget | IASOWidget | None = None,
@@ -66,14 +67,18 @@ def __init__(
         if choices is not None:
             if not self.type.accepts_choices:
                 raise InvalidParameterError(f"Parameters of type {self.type} don't accept choices.")
-            elif len(choices) == 0:
-                raise InvalidParameterError("Choices, if provided, cannot be empty.")
-
-            try:
-                for choice in choices:
-                    self.type.validate(choice)
-            except ParameterValueError:
-                raise InvalidParameterError(f"The provided choices are not valid for the {self.type} parameter type.")
+            if isinstance(choices, str):
+                choices = ChoicesFromFile(choices)
+            elif not isinstance(choices, ChoicesFromFile):
+                if len(choices) == 0:
+                    raise InvalidParameterError("Choices, if provided, cannot be empty.")
+                try:
+                    for choice in choices:
+                        self.type.validate(choice)
+                except ParameterValueError:
+                    raise InvalidParameterError(
+                        f"The provided choices are not valid for the {self.type} parameter type."
+                    )
         self.choices = choices
 
         self.name = name
@@ -100,11 +105,11 @@ def validate(self, value: typing.Any) -> typing.Any:
 
     def to_dict(self) -> dict[str, typing.Any]:
         """Return a dictionary representation of the Parameter instance."""
-        return {
+        d = {
             "code": self.code,
             "type": self.type.spec_type,
             "name": self.name,
-            "choices": self.choices,
+            "choices": None if isinstance(self.choices, ChoicesFromFile) else self.choices,
             "help": self.help,
             "default": self.default,
             "widget": self.widget.value if self.widget else None,
@@ -113,6 +118,9 @@ def to_dict(self) -> dict[str, typing.Any]:
             "multiple": self.multiple,
             "directory": self.directory,
         }
+        if isinstance(self.choices, ChoicesFromFile):
+            d["choices_from_file"] = self.choices.to_dict()
+        return d
 
     def _validate_single(self, value: typing.Any):
         # Normalize empty values to None and handles default
@@ -129,7 +137,11 @@ def _validate_single(self, value: typing.Any):
                 return None
 
         pre_validated = self.type.validate(normalized_value)
-        if self.choices is not None and pre_validated not in self.choices:
+        if (
+            self.choices is not None
+            and not isinstance(self.choices, ChoicesFromFile)
+            and pre_validated not in self.choices
+        ):
             raise ParameterValueError(f"The provided value for {self.code} is not included in the provided choices.")
 
         return pre_validated
@@ -152,7 +164,11 @@ def _validate_multiple(self, value: typing.Any):
             raise ParameterValueError(f"{self.code} is required")
 
         pre_validated = [self.type.validate(single_value) for single_value in normalized_value]
-        if self.choices is not None and any(v not in self.choices for v in pre_validated):
+        if (
+            self.choices is not None
+            and not isinstance(self.choices, ChoicesFromFile)
+            and any(v not in self.choices for v in pre_validated)
+        ):
             raise ParameterValueError(
                 f"One of the provided values for {self.code} is not included in the provided choices."
             )
@@ -174,7 +190,7 @@ def _validate_default(self, default: typing.Any, multiple: bool):
         except ParameterValueError:
             raise InvalidParameterError(f"The default value for {self.code} is not valid.")
 
-        if self.choices is not None:
+        if self.choices is not None and not isinstance(self.choices, ChoicesFromFile):
             if isinstance(default, list):
                 if not all(d in self.choices for d in default):
                     raise InvalidParameterError(
@@ -227,7 +243,7 @@ def parameter(
         | File
     ],
     name: str | None = None,
-    choices: typing.Sequence | None = None,
+    choices: typing.Sequence | ChoicesFromFile | str | None = None,
     help: str | None = None,
     widget: DHIS2Widget | IASOWidget | None = None,
     connection: str | None = None,
@@ -261,7 +277,7 @@ def parameter(
         An optional default value for the parameter (should be of the type defined by the type parameter)
     required : bool, default=True
         Whether the parameter is mandatory
-    multiple : bool, default=True
+    multiple : bool, default=False
         Whether this parameter should be provided multiple values (if True, the value must be provided as a list of
         values of the chosen type)
     directory : str, optional

openhexa/sdk/pipelines/runtime.py

@@ -6,6 +6,7 @@
 import io
 import os
 import sys
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -16,6 +17,7 @@
 from openhexa.sdk.pipelines.exceptions import InvalidParameterError, PipelineNotFound
 from openhexa.sdk.pipelines.parameter import (
     TYPES_BY_PYTHON_TYPE,
+    ChoicesFromFile,
     DHIS2Widget,
     IASOWidget,
     Parameter,
@@ -25,6 +27,12 @@
 
 from .pipeline import Pipeline
 
+# Maps AST function names to classes that support from_ast_call().
+# Add an entry here when introducing a new AstConstructible type.
+_AST_CALLABLE_TYPES: dict[str, type] = {
+    "ChoicesFromFile": ChoicesFromFile,
+}
+
 
 @dataclass
 class Argument:
@@ -33,6 +41,7 @@ class Argument:
     name: str  # Use str instead of string
     types: list[type] = field(default_factory=list)
     default_value: Any = None
+    transform: Callable | None = None
 
 
 def import_pipeline(pipeline_dir_path: str) -> Pipeline:
@@ -172,6 +181,12 @@ def _get_decorator_arg_value(decorator: ast.Call, arg: Argument, index: int) ->
                 return (keyword.value.id, True)
             elif isinstance(keyword.value, ast.List):
                 return ([el.value for el in keyword.value.elts], True)
+            elif isinstance(keyword.value, ast.Call):
+                func = keyword.value.func
+                func_name = func.id if isinstance(func, ast.Name) else None
+                if func_name not in _AST_CALLABLE_TYPES:
+                    raise ValueError(f"Unsupported call in choices argument: {func_name}")
+                return _AST_CALLABLE_TYPES[func_name].from_ast_call(keyword.value), True
             elif isinstance(keyword.value, ast.Attribute):
                 if keyword.value.attr in DHIS2Widget.__members__:
                     return getattr(DHIS2Widget, keyword.value.attr), True
@@ -201,6 +216,8 @@ def _get_decorator_spec(decorator: ast.Call, args: tuple[Argument, ...]) -> dict
     args_spec = {}
     for i, arg in enumerate(args):
         value, is_keyword = _get_decorator_arg_value(decorator, arg, i)
+        if arg.transform is not None:
+            value = arg.transform(value)
         args_spec[arg.name] = {"value": value, "is_keyword": is_keyword}
     return args_spec
 
@@ -287,7 +304,11 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
                     Argument("code", [ast.Constant]),
                     Argument("type", [ast.Name]),
                     Argument("name", [ast.Constant]),
-                    Argument("choices", [ast.List]),
+                    Argument(
+                        "choices",
+                        [ast.List, ast.Call, ast.Constant],
+                        transform=lambda v: ChoicesFromFile(v) if isinstance(v, str) else v,
+                    ),
                     Argument("help", [ast.Constant]),
                     Argument("default", [ast.Constant, ast.List]),
                     Argument("widget", [ast.Attribute]),