feat: minor improvements

Author: mrivar

openhexa/sdk/pipelines/parameter/ast_constructible.py

@@ -8,9 +8,14 @@ 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 to gain automatic ``from_ast_call`` support in
-    the pipeline AST parser — no changes to the parser needed when new parameters
-    are added to the class.
+    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
@@ -21,10 +26,19 @@ def from_ast_call(cls, node: ast.Call) -> "AstConstructible":
         ``inspect.signature``, then merges keyword args, and calls ``cls``.
         """
         param_names = list(inspect.signature(cls).parameters.keys())
-        kwargs = {
-            param_names[i]: arg.value
-            for i, arg in enumerate(node.args)
-            if isinstance(arg, ast.Constant) and i < len(param_names)
-        }
-        kwargs |= {kw.arg: kw.value.value for kw in node.keywords if isinstance(kw.value, ast.Constant)}
+        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

@@ -26,13 +26,11 @@ class ChoicesFromFile(AstConstructible):
     def __init__(self, path: str, column: str | None = None):
         self.path = path
         self.column = column
-        self.format = self._detect_format(path)
         self.validate_spec()
+        self.format = self._detect_format(path)
 
     @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(
@@ -48,6 +46,19 @@ def validate_spec(self):
         if self.column is not None and not isinstance(self.column, str):
             raise InvalidParameterError("ChoicesFromFile column must be a string.")
 
+    def __repr__(self) -> str:
+        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:
+        if not isinstance(other, ChoicesFromFile):
+            return NotImplemented
+        return self.path == other.path and self.column == other.column
+
+    def __hash__(self) -> int:
+        return hash((self.path, self.column))
+
     def to_dict(self) -> dict:
         """Return a dictionary representation of the choices spec."""
         return {

openhexa/sdk/pipelines/parameter/decorator.py

@@ -69,10 +69,7 @@ def __init__(
                 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
-            else:
+            elif not isinstance(choices, ChoicesFromFile):
                 if len(choices) == 0:
                     raise InvalidParameterError("Choices, if provided, cannot be empty.")
                 try:

openhexa/sdk/pipelines/runtime.py

@@ -23,15 +23,15 @@
     Parameter,
     validate_parameters,
 )
+from openhexa.sdk.utils import Settings
+
+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,
 }
-from openhexa.sdk.utils import Settings
-
-from .pipeline import Pipeline
 
 
 @dataclass