feature: dynamic choices using FileChoices - first iteration

Author: mrivar

openhexa/sdk/pipelines/parameter/__init__.py

@@ -5,6 +5,7 @@
 
 from openhexa.sdk.pipelines.exceptions import InvalidParameterError, ParameterValueError
 
+from .choices import FileChoices
 from .decorator import FunctionWithParameter, Parameter, parameter, validate_parameters
 from .types import (
     Boolean,
@@ -56,6 +57,8 @@
     "SecretType",
     # Registry
     "TYPES_BY_PYTHON_TYPE",
+    # Dynamic choices
+    "FileChoices",
     # Widgets
     "DHIS2Widget",
     "IASOWidget",

openhexa/sdk/pipelines/parameter/choices.py

@@ -0,0 +1,52 @@
+"""Dynamic choices classes for pipeline parameters."""
+
+from openhexa.sdk.pipelines.exceptions import InvalidParameterError
+
+_SUPPORTED_FORMATS = {"csv", "json", "yaml", "yml"}
+
+
+class FileChoices:
+    """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()
+
+    def _detect_format(self, path: str) -> str:
+        if not path or not isinstance(path, str):
+            raise InvalidParameterError("FileChoices 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):
+        if not self.path or not isinstance(self.path, str):
+            raise InvalidParameterError("FileChoices path must be a non-empty string.")
+        if self.column is not None and not isinstance(self.column, str):
+            raise InvalidParameterError("FileChoices column must be a string.")
+
+    def to_dict(self) -> dict:
+        return {
+            "format": self.format,
+            "path": self.path,
+            "column": self.column,
+        }

openhexa/sdk/pipelines/parameter/decorator.py

@@ -15,6 +15,7 @@
     S3Connection,
 )
 
+from .choices import FileChoices
 from .types import Boolean, DHIS2ConnectionType, IASOConnectionType, Secret, TYPES_BY_PYTHON_TYPE
 from .widgets import DHIS2Widget, IASOWidget
 
@@ -42,7 +43,7 @@ def __init__(
             | File
         ],
         name: str | None = None,
-        choices: typing.Sequence | None = None,
+        choices: typing.Sequence | FileChoices | None = None,
         help: str | None = None,
         default: typing.Any | None = None,
         widget: DHIS2Widget | IASOWidget | None = None,
@@ -66,14 +67,19 @@ 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, FileChoices):
+                # validate_spec() already ran in FileChoices.__init__; nothing more to check here
+                pass
+            else:
+                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 +106,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, FileChoices) else self.choices,
             "help": self.help,
             "default": self.default,
             "widget": self.widget.value if self.widget else None,
@@ -113,6 +119,9 @@ def to_dict(self) -> dict[str, typing.Any]:
             "multiple": self.multiple,
             "directory": self.directory,
         }
+        if isinstance(self.choices, FileChoices):
+            d["file_choices"] = self.choices.to_dict()
+        return d
 
     def _validate_single(self, value: typing.Any):
         # Normalize empty values to None and handles default
@@ -129,7 +138,7 @@ 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, FileChoices) 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 +161,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, FileChoices)
+            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 +187,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, FileChoices):
             if isinstance(default, list):
                 if not all(d in self.choices for d in default):
                     raise InvalidParameterError(
@@ -227,7 +240,7 @@ def parameter(
         | File
     ],
     name: str | None = None,
-    choices: typing.Sequence | None = None,
+    choices: typing.Sequence | FileChoices | None = None,
     help: str | None = None,
     widget: DHIS2Widget | IASOWidget | None = None,
     connection: str | None = None,

openhexa/sdk/pipelines/runtime.py

@@ -17,6 +17,7 @@
 from openhexa.sdk.pipelines.parameter import (
     TYPES_BY_PYTHON_TYPE,
     DHIS2Widget,
+    FileChoices,
     IASOWidget,
     Parameter,
     validate_parameters,
@@ -172,6 +173,21 @@ 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 != "FileChoices":
+                    raise ValueError(f"Unsupported call in choices argument: {func_name}")
+                # Extract positional arg (path) and keyword args (column, format override)
+                pos_args = [a.value for a in keyword.value.args if isinstance(a, ast.Constant)]
+                kw_args = {
+                    kw.arg: kw.value.value
+                    for kw in keyword.value.keywords
+                    if isinstance(kw.value, ast.Constant)
+                }
+                if pos_args:
+                    kw_args.setdefault("path", pos_args[0])
+                return (FileChoices(**kw_args), True)
             elif isinstance(keyword.value, ast.Attribute):
                 if keyword.value.attr in DHIS2Widget.__members__:
                     return getattr(DHIS2Widget, keyword.value.attr), True
@@ -287,7 +303,7 @@ 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]),
                     Argument("help", [ast.Constant]),
                     Argument("default", [ast.Constant, ast.List]),
                     Argument("widget", [ast.Attribute]),