Merge pull request #3 from flusflas/feature/dynamic-expressions

Adds `$eval()` for evaluating complex expressions.

Author: flusflas

CHANGELOG.md

@@ -2,8 +2,33 @@
 
 All notable changes to this project will be documented in this file.
 
-The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
-and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+> ⚠️ **Warning: Pre-1.0.0 Release**
+> 
+> This project is in early development. Breaking changes may occur in any minor
+> or patch release until version 1.0.0. Follow changelogs closely and pin
+> versions if needed.
+
+## [0.4.0](https://github.com/flusflas/apier/tree/v0.4.0) (2025-08-04)
+
+This release adds `$eval` support for dynamic expressions, allowing complex
+evaluations and manipulations of request and response data, and unlocking
+advanced pagination strategies.
+
+### 🧨 Breaking Changes
+
+- Renamed `reuse-previous-request` to `reuse_previous_request` in pagination
+  configuration for naming consistency.
+- Enforced `#` prefix for dot-separated paths in dynamic expressions (e.g.,
+  `#users.0.name` instead of `users.0.name`).
+
+### Added
+
+- Simple expression evaluation with `ast` module.
+- Support for nullish coalescing operator (`??`) in dynamic expressions.
+
+### Fixed
+
+- Evaluation of `results` parameter in pagination configuration.
 
 ## [0.3.0](https://github.com/flusflas/apier/tree/v0.3.0) (2025-07-31)
 

README.md

@@ -13,6 +13,8 @@
 
 apier provides a command-line interface (CLI) for building client libraries and merging OpenAPI documents.
 
+> 🐣 This project is in pre-1.0.0. Expect breaking changes in minor and patch versions.
+
 ## 🐍 Installation
 
 apier is available on PyPI:

apier/extensions/pagination.py

@@ -9,7 +9,7 @@ class PaginationDescription(BaseModel):
     class Config:
         allow_population_by_field_name = True
 
-    reuse_previous_request: bool = Field(default=False, alias="reuse-previous-request")
+    reuse_previous_request: bool = Field(default=False, alias="reuse_previous_request")
     method: str = ""
     url: str = ""
     modifiers: List[PaginationModifier] = Field(default_factory=list)
@@ -20,13 +20,13 @@ class Config:
     def validate_fields(cls, values: dict):
         if isinstance(values, PaginationDescription):
             values = values.dict()
-        reuse = values.get("reuse-previous-request") or values.get(
+        reuse = values.get("reuse_previous_request") or values.get(
             "reuse_previous_request"
         )
         for attr in ["method", "url"]:
             if not reuse and not values[attr]:
                 raise ValueError(
-                    f"The field '{attr}' is required if 'reuse-previous-request' is False"
+                    f"The field '{attr}' is required if 'reuse_previous_request' is False"
                 )
         return values
 

apier/templates/python_tree/base/internal/expressions/__init__.py

@@ -0,0 +1,125 @@
+import ast
+import operator
+
+# Supported binary operators
+allowed_operators = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+    ast.USub: operator.neg,  # Unary minus
+}
+
+# Supported comparison operators
+allowed_comparators = {
+    ast.Eq: operator.eq,
+    ast.NotEq: operator.ne,
+    ast.Lt: operator.lt,
+    ast.LtE: operator.le,
+    ast.Gt: operator.gt,
+    ast.GtE: operator.ge,
+}
+
+# Whitelisted functions
+allowed_functions = {
+    "int": int,
+    "len": len,
+    "round": lambda x: int(round(x)),
+    "floor": lambda x: int(x // 1),
+    "ceil": lambda x: int(-(-x // 1)),
+}
+
+# Constants
+allowed_constants = {
+    "true": True,
+    "false": False,
+    "null": None,
+}
+
+
+def eval_expr(expr, vars=None):
+    """
+    Evaluates a compound expression using AST with a restricted subset of
+    Python syntax:
+     - Arithmetic (+, -, *, /, unary -)
+     - Function calls (`int`, `len`, `round`, `floor`, `ceil`)
+     - Comparisons (==, !=, <, <=, >, >=)
+
+    Note: This implementation is a basic and temporary solution based on Python syntax.
+    It has limited extensibility and is not designed to be portable across other languages.
+    For more complex use cases, a dedicated expression language or library would be required.
+
+    Parameters:
+        expr (str): The expression string to evaluate.
+        vars (dict, optional): A dictionary of variables to use in the expression.
+
+    Returns:
+        The result of evaluating the expression.
+    """
+    # Parse the expression into an AST
+    try:
+        tree = ast.parse(expr, mode="eval")
+    except SyntaxError as e:
+        raise SyntaxError(f"Invalid expression syntax: {e.msg}") from e
+
+    vars = vars.copy() if vars is not None else {}
+    vars.update(allowed_constants)
+
+    def _eval(node: ast.AST):
+        """Recursively evaluate supported AST nodes."""
+
+        if isinstance(node, ast.Constant):
+            if isinstance(node.value, complex):
+                raise ValueError("Complex numbers are not supported.")
+            if node.value is True or node.value is False or node.value is None:
+                raise ValueError(f"Unsupported constant: {node.value}")
+            return node.value
+
+        elif isinstance(node, ast.Name):
+            # Variable reference
+            if vars is not None and node.id in vars:
+                return vars[node.id]
+            else:
+                raise ValueError(f"Variable not defined: {node.id}")
+
+        elif isinstance(node, ast.BinOp) and type(node.op) in allowed_operators:
+            left = _eval(node.left)
+            right = _eval(node.right)
+            return allowed_operators[type(node.op)](left, right)
+
+        elif isinstance(node, ast.UnaryOp) and type(node.op) in allowed_operators:
+            operand = _eval(node.operand)
+            return allowed_operators[type(node.op)](operand)
+
+        elif isinstance(node, ast.Call):
+            # Only allow calls to whitelisted functions
+            if isinstance(node.func, ast.Name) and node.func.id in allowed_functions:
+                func = allowed_functions[node.func.id]
+                args = [_eval(arg) for arg in node.args]
+                return func(*args)
+            else:
+                raise ValueError(f"Function not allowed: '{node.func.id}'")
+
+        elif isinstance(node, ast.List):
+            return [_eval(elt) for elt in node.elts]
+
+        elif isinstance(node, ast.Tuple):
+            return tuple(_eval(elt) for elt in node.elts)
+
+        elif isinstance(node, ast.Compare):
+            # Only support simple comparisons (not chained comparisons)
+            if len(node.ops) != 1 or len(node.comparators) != 1:
+                raise ValueError("Only simple comparisons are supported.")
+            op = node.ops[0]
+            if type(op) not in allowed_comparators:
+                raise ValueError(
+                    f"Comparison operator not allowed: {type(op).__name__}"
+                )
+            left = _eval(node.left)
+            right = _eval(node.comparators[0])
+            return allowed_comparators[type(op)](left, right)
+
+        else:
+            raise SyntaxError(f"Unsupported syntax: {type(node).__name__}")
+
+    return _eval(tree.body)

apier/templates/python_tree/base/internal/expressions/evaluation.py

@@ -6,6 +6,8 @@
 
 from requests import PreparedRequest, Request, Response
 
+from .evaluation import eval_expr
+
 
 class RuntimeExpressionError(Exception):
     """Invalid runtime expression."""
@@ -21,6 +23,7 @@ def evaluate(
     path_values: dict = None,
     query_param_types: dict = None,
     header_param_types: dict = None,
+    **kwargs,
 ):
     """
     Evaluates an OpenAPI runtime expression (https://swagger.io/docs/specification/links/).
@@ -45,23 +48,85 @@ def evaluate(
                         will be returned.
     :return:            The result of the evaluated expression.
     """
+
+    # The 'eval_variables'  is a dictionary used to store variables when
+    # evaluating expressions within an $eval() expression. The result of the
+    # evaluation will be stored in this dictionary. This parameter is intended
+    # for internal use only and is not meant to be accessed directly outside the
+    # function.
+    eval_variables = kwargs.get("eval_variables")
+    is_eval = eval_variables is not None
+
     try:
         expression = expression.strip()
 
+        if expression.startswith("$eval("):
+            if is_eval:
+                raise RuntimeExpressionError(
+                    caused_by=ValueError(
+                        "Nested evaluation expressions are not supported"
+                    )
+                )
+
+            inner_expression = expression[len("$eval(") : -1].strip()
+            eval_variables = {}
+            expr = evaluate(
+                resp,
+                inner_expression,
+                path_values=path_values,
+                query_param_types=query_param_types,
+                header_param_types=header_param_types,
+                eval_variables=eval_variables,
+            )
+
+            return eval_expr(expr, eval_variables)
+
         if "{" in expression:
 
-            def replace_group(match):
-                return str(
-                    _evaluate_runtime_expression(
+            def replace_match(match):
+                default_value_defined = " ?? " in match
+
+                # Handle coalescing operator (??) to provide a default value
+                if default_value_defined:
+                    match, default_value = map(str.strip, match.split(" ?? ", 1))
+                    default_value = eval_expr(default_value)
+
+                try:
+                    group_result = evaluate(
                         resp,
-                        match.group(1),
+                        match,
                         path_values=path_values,
                         query_param_types=query_param_types,
                         header_param_types=header_param_types,
                     )
-                )
-
-            return re.sub(r"{(\$[^}]+)}", replace_group, expression)
+                except RuntimeExpressionError as e:
+                    if default_value_defined and isinstance(
+                        e.caused_by, (KeyError, IndexError)
+                    ):
+                        group_result = default_value
+                    else:
+                        raise e
+
+                # If inside an $eval(), the expression must be replaced with
+                # variables instead of the group result
+                if is_eval:
+                    var_name = f"var{len(eval_variables)}"
+                    eval_variables[var_name] = group_result
+                    return var_name
+
+                return group_result
+
+            parts = re.split(r"({\$?[^}]+})", expression)
+            parts = [part for part in parts if part]
+            for i, part in enumerate(parts):
+                if part.startswith("{") and part.endswith("}"):
+                    parts[i] = replace_match(part[1:-1])
+            return parts[0] if len(parts) == 1 else "".join(map(str, parts))
+
+        # If inside an $eval() expression, any subexpressions (enclosed in {})
+        # should already have been resolved, so the expression can be returned as is
+        if is_eval:
+            return expression
 
         if expression.startswith("$"):
             return _evaluate_runtime_expression(
@@ -72,15 +137,22 @@ def replace_group(match):
                 header_param_types=header_param_types,
             )
 
-        if isinstance(resp, Response):
-            resp = resp.json()
+        if expression.startswith("#"):
+            # Dot-separated path
+            if isinstance(resp, Response):
+                resp = resp.json()
+
+            if not isinstance(resp, dict):
+                raise ValueError("Invalid response format")
 
-        if not isinstance(resp, dict):
-            raise ValueError("Invalid dict response")
+            return _get_from_dict(resp, expression[1:].strip())
+
+        # Return the expression as a literal value
+        return expression
 
-        return _get_from_dict(resp, expression)
     except RuntimeExpressionError as e:
         raise e
+
     except Exception as e:
         raise RuntimeExpressionError(caused_by=e)
 
@@ -112,7 +184,9 @@ def get_query_string(name):
         query_params = parse_qs(parsed_url.query)
         value = query_params.get(name, [])
         if len(value) == 0:
-            raise RuntimeExpressionError(f"Query parameter '{name}' not found")
+            raise RuntimeExpressionError(
+                caused_by=KeyError(f"Query parameter '{name}' not found")
+            )
         elif len(value) == 1:
             return value[0]
         else:
@@ -121,7 +195,9 @@ def get_query_string(name):
     def get_path_value(name):
         if path_values is not None and name in path_values:
             return path_values[name]
-        raise RuntimeExpressionError(f"Path parameter '{name}' not found")
+        raise RuntimeExpressionError(
+            caused_by=KeyError(f"Path parameter '{name}' not found")
+        )
 
     expression_funcs = {
         "$url": lambda: resp.request.url,
@@ -151,7 +227,7 @@ def get_path_value(name):
         if expr == expression:
             return fn()
 
-    raise RuntimeExpressionError("invalid runtime expression")
+    raise RuntimeExpressionError(caused_by=ValueError("Invalid runtime expression"))
 
 
 def prepare_request(req: Union[PreparedRequest, Request], expression: str, value):
@@ -237,6 +313,9 @@ def set_query_param(name):
 
 
 def _get_from_dict(d: dict, key: str, separator="."):
+    if key == "":
+        return d
+
     try:
 
         def get_item(a, b):

…ython_tree/base/internal/runtime_expr.py …ree/base/internal/expressions/runtime.pyapier/templates/python_tree/base/internal/runtime_expr.py renamed to apier/templates/python_tree/base/internal/expressions/runtime.py apier/templates/python_tree/base/internal/runtime_expr.py renamed to apier/templates/python_tree/base/internal/expressions/runtime.py

@@ -21,7 +21,7 @@
     content_types_compatible,
     ContentTypeValidationResult,
 )
-from .runtime_expr import evaluate, prepare_request
+from .expressions.runtime import evaluate, prepare_request
 from ..models.basemodel import APIBaseModel
 from ..models.exceptions import ExceptionList, ResponseError
 from ..models.extensions.pagination import PaginationDescription
@@ -269,10 +269,8 @@ def _handle_pagination(
             )
             req.prepare_url(url, None)
         if pagination_info.method:
-            # TODO: Evaluate expression ???
             req.prepare_method(pagination_info.method)
         for modifier in pagination_info.modifiers:
-            # TODO: Evaluate complex expressions
             value = evaluate(resp, modifier.value, path_values, query_params, headers)
             prepare_request(req, modifier.param, value)