feat: add string shorthand choices

Author: mrivar

openhexa/sdk/pipelines/parameter/decorator.py

@@ -43,7 +43,7 @@ def __init__(
             | File
         ],
         name: str | None = None,
-        choices: typing.Sequence | ChoicesFromFile | None = None,
+        choices: typing.Sequence | ChoicesFromFile | str | None = None,
         help: str | None = None,
         default: typing.Any | None = None,
         widget: DHIS2Widget | IASOWidget | None = None,
@@ -67,6 +67,8 @@ 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.")
+            if isinstance(choices, str):
+                choices = ChoicesFromFile(choices)
             if isinstance(choices, ChoicesFromFile):
                 # validate_spec() already ran in ChoicesFromFile.__init__; nothing more to check here
                 pass
@@ -244,7 +246,7 @@ def parameter(
         | File
     ],
     name: str | None = None,
-    choices: typing.Sequence | ChoicesFromFile | None = None,
+    choices: typing.Sequence | ChoicesFromFile | str | None = None,
     help: str | None = None,
     widget: DHIS2Widget | IASOWidget | None = None,
     connection: str | None = None,

openhexa/sdk/pipelines/runtime.py

@@ -8,7 +8,7 @@
 import sys
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 from zipfile import ZipFile
 
 import requests
@@ -37,9 +37,10 @@
 class Argument:
     """Argument of a decorator."""
 
-    name: str  # Use str instead of string
+    name: str
     types: list[type] = field(default_factory=list)
     default_value: Any = None
+    transform: Callable | None = None
 
 
 def import_pipeline(pipeline_dir_path: str) -> Pipeline:
@@ -214,6 +215,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
 
@@ -300,7 +303,11 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
                     Argument("code", [ast.Constant]),
                     Argument("type", [ast.Name]),
                     Argument("name", [ast.Constant]),
-                    Argument("choices", [ast.List, ast.Call]),
+                    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]),

tests/test_choices.py

@@ -57,6 +57,196 @@ def test_to_dict_no_column(self):
         assert fc.to_dict() == {"format": "csv", "path": "districts.csv", "column": None}
 
 
+# ---------------------------------------------------------------------------
+# String shorthand — Parameter.__init__
+# ---------------------------------------------------------------------------
+
+
+class TestStringShorthand:
+    # --- happy paths ---
+
+    def test_string_shorthand_csv(self):
+        p = Parameter(code="district", type=str, choices="districts.csv")
+        assert isinstance(p.choices, ChoicesFromFile)
+        assert p.choices.path == "districts.csv"
+        assert p.choices.format == "csv"
+        assert p.choices.column is None
+
+    def test_string_shorthand_json(self):
+        p = Parameter(code="district", type=str, choices="data/regions.json")
+        assert isinstance(p.choices, ChoicesFromFile)
+        assert p.choices.format == "json"
+
+    def test_string_shorthand_yaml(self):
+        p = Parameter(code="district", type=str, choices="list.yaml")
+        assert isinstance(p.choices, ChoicesFromFile)
+        assert p.choices.format == "yaml"
+
+    def test_string_shorthand_yml(self):
+        p = Parameter(code="district", type=str, choices="list.yml")
+        assert isinstance(p.choices, ChoicesFromFile)
+        assert p.choices.format == "yaml"
+
+    def test_string_shorthand_leading_slash_stripped(self):
+        p = Parameter(code="district", type=str, choices="/choices.csv")
+        assert p.choices.path == "/choices.csv"  # ChoicesFromFile stores as-is; stripping is app-side
+
+    def test_string_shorthand_serialises_same_as_explicit(self):
+        shorthand = Parameter(code="district", type=str, choices="districts.csv").to_dict()
+        explicit = Parameter(code="district", type=str, choices=ChoicesFromFile("districts.csv")).to_dict()
+        assert shorthand == explicit
+
+    # --- static list still works ---
+
+    def test_static_list_unaffected(self):
+        p = Parameter(code="country", type=str, choices=["UG", "KE"])
+        assert p.choices == ["UG", "KE"]
+
+    def test_explicit_choices_from_file_unaffected(self):
+        p = Parameter(code="district", type=str, choices=ChoicesFromFile("districts.csv", column="code"))
+        assert p.choices.column == "code"
+
+    # --- invalid strings raise clearly ---
+
+    def test_string_no_extension_raises(self):
+        with pytest.raises(InvalidParameterError, match="Supported extensions"):
+            Parameter(code="district", type=str, choices="nodot")
+
+    def test_string_unsupported_extension_raises(self):
+        with pytest.raises(InvalidParameterError, match="Supported extensions"):
+            Parameter(code="district", type=str, choices="file.xlsx")
+
+    def test_empty_string_raises(self):
+        with pytest.raises(InvalidParameterError):
+            Parameter(code="district", type=str, choices="")
+
+    # --- column cannot be specified via shorthand ---
+
+    def test_shorthand_has_no_column(self):
+        p = Parameter(code="district", type=str, choices="districts.csv")
+        assert p.choices.column is None
+
+    def test_decorator_with_string_shorthand(self):
+        @parameter(code="district", type=str, choices="districts.csv")
+        def my_pipeline(district):
+            pass
+
+        params = my_pipeline.get_all_parameters()
+        assert isinstance(params[0].choices, ChoicesFromFile)
+
+
+# ---------------------------------------------------------------------------
+# String shorthand — AST round-trip
+# ---------------------------------------------------------------------------
+
+
+class TestAstStringShorthand(TestCase):
+    def _write_pipeline(self, tmpdir, param_line):
+        with open(f"{tmpdir}/pipeline.py", "w") as f:
+            f.write(
+                "\n".join(
+                    [
+                        "from openhexa.sdk.pipelines import pipeline, parameter",
+                        "",
+                        param_line,
+                        "@pipeline(name='Test pipeline')",
+                        "def test_pipeline(district):",
+                        "    pass",
+                    ]
+                )
+            )
+
+    def test_ast_string_shorthand_csv(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices='districts.csv')",
+            )
+            p = get_pipeline(tmpdir)
+            param_dict = p.to_dict()["parameters"][0]
+            assert param_dict["choices"] is None
+            assert param_dict["choices_from_file"] == {"format": "csv", "path": "districts.csv", "column": None}
+
+    def test_ast_string_shorthand_json(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices='regions.json')",
+            )
+            p = get_pipeline(tmpdir)
+            assert p.to_dict()["parameters"][0]["choices_from_file"]["format"] == "json"
+
+    def test_ast_string_shorthand_yaml(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices='list.yml')",
+            )
+            p = get_pipeline(tmpdir)
+            assert p.to_dict()["parameters"][0]["choices_from_file"]["format"] == "yaml"
+
+    def test_ast_string_shorthand_same_output_as_explicit(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices='districts.csv')",
+            )
+            shorthand_dict = get_pipeline(tmpdir).to_dict()["parameters"][0]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices=ChoicesFromFile('districts.csv'))",
+            )
+            # need the import for the explicit form
+            with open(f"{tmpdir}/pipeline.py", "w") as f:
+                f.write(
+                    "\n".join(
+                        [
+                            "from openhexa.sdk.pipelines import pipeline, parameter",
+                            "from openhexa.sdk.pipelines.parameter import ChoicesFromFile",
+                            "",
+                            "@parameter('district', type=str, choices=ChoicesFromFile('districts.csv'))",
+                            "@pipeline(name='Test pipeline')",
+                            "def test_pipeline(district):",
+                            "    pass",
+                        ]
+                    )
+                )
+            explicit_dict = get_pipeline(tmpdir).to_dict()["parameters"][0]
+
+        assert shorthand_dict == explicit_dict
+
+    def test_ast_static_list_unaffected(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('country', type=str, choices=['UG', 'KE'])",
+            )
+            p = get_pipeline(tmpdir)
+            param_dict = p.to_dict()["parameters"][0]
+            assert param_dict["choices"] == ["UG", "KE"]
+            assert "choices_from_file" not in param_dict
+
+    def test_ast_string_no_extension_raises(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices='nodot')",
+            )
+            with self.assertRaises(InvalidParameterError):
+                get_pipeline(tmpdir)
+
+    def test_ast_string_unsupported_extension_raises(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self._write_pipeline(
+                tmpdir,
+                "@parameter('district', type=str, choices='file.xlsx')",
+            )
+            with self.assertRaises(InvalidParameterError):
+                get_pipeline(tmpdir)
+
+
 # ---------------------------------------------------------------------------
 # Parameter integration
 # ---------------------------------------------------------------------------