refactor: improve pagination handling and input validation

Ensure pagination modifiers and parameters are nullable, simplify case
handling with utility method, and improve mutual exclusivity validation.
Updated tests and templates to reflect the changes, including consistent
snake_case for operation IDs.

Author: flusflas

apier/extensions/pagination.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import List, Optional
+from typing import List, Optional, Callable
 
 from pydantic import BaseModel, Field, root_validator
 
@@ -42,19 +42,19 @@ class Config:
         description="The operation ID of the next API operation, as defined in the OpenAPI spec.",
     )
     parameters: Optional[List[NextOperationParameters]] = Field(
-        [],
+        None,
         description="Parameters to be passed to the next operation. If required "
         "parameters are not provided, such as path parameters, the operation "
         "will fail.",
     )
 
     # Modifiers mode
-    reuse_previous_request: bool = Field(
-        default=False,
+    reuse_previous_request: Optional[bool] = Field(
+        None,
         description="Whether the next request should reuse the previous request's parameters.",
     )
-    modifiers: List[PaginationModifier] = Field(
-        default_factory=list,
+    modifiers: Optional[List[PaginationModifier]] = Field(
+        None,
         description="List of request modifiers to update parameters for the next request.",
     )
 
@@ -68,28 +68,45 @@ class Config:
         description="A dynamic expression that evaluates to a boolean indicating if there are more results.",
     )
 
-    @root_validator
+    @root_validator(pre=True)
     def check_mutually_exclusive_fields(cls, values):
         """
-        Validates that the modes are mutually exclusive.
-         - If either `operation_id` or `parameters` is set, use 'operation' mode.
-           Otherwise, use 'modifiers' mode.
-         - In 'operation' mode, `operation_id` must be present.
+        Enforces mutual exclusivity between:
+         - operation mode: operation_id and/or parameters
+         - modifiers mode: reuse_previous_request and/or modifiers
+
+        In 'operation' mode, `operation_id` must be present.
         """
 
+        if ("operation_id" in values or "parameters" in values) and (
+            "reuse_previous_request" in values or "modifiers" in values
+        ):
+            raise ValueError(
+                "Cannot use 'operation_id' or 'parameters' with 'reuse_previous_request' or 'modifiers'."
+            )
+
         operation_id = values.get("operation_id")
         operation_params = values.get("parameters")
 
         mode = "operation" if operation_id or operation_params else "modifiers"
         if mode == "operation" and not operation_id:
             raise ValueError("'operation_id' is required in operation mode.")
 
-        if mode == "operation" and not operation_id:
-            raise ValueError(
-                "'operation_id' is required if 'operation_params' are set."
-            )
-
         return values
 
+    def update_operation_case(
+        self, func: Callable[[str], str]
+    ) -> PaginationDescription:
+        """
+        Updates the case of the operation ID and parameters using the provided function.
+        """
+        if self.operation_id:
+            self.operation_id = func(self.operation_id)
+
+        for param in self.parameters or []:
+            param.name = func(param.name)
+
+        return self
+
 
 PaginationDescription.update_forward_refs()

apier/templates/python_tree/base/internal/resource.py

@@ -267,7 +267,7 @@ def _handle_pagination(
         if pagination_info.reuse_previous_request:
             req = resp.request
 
-        for modifier in pagination_info.modifiers:
+        for modifier in pagination_info.modifiers or []:
             value = evaluate(resp, modifier.value, path_values, query_params, headers)
             prepare_request(req, modifier.param, value)
 
@@ -278,7 +278,7 @@ def _handle_pagination(
             op_id = pagination_info.operation_id
             evaluated_params = {}
 
-            for param in pagination_info.parameters:
+            for param in pagination_info.parameters or []:
                 evaluated_params[param.name] = evaluate(
                     resp, param.value, path_values, query_params, headers
                 )
@@ -292,10 +292,6 @@ def _handle_pagination(
 
             next_op = getattr(api_operations, op_id)
 
-            # Check if next_op accepts a "req" parameter
-            if "req" in next_op.__code__.co_varnames:
-                evaluated_params["req"] = req.body
-
             ret._pagination.iter_func = lambda: next_op(**evaluated_params)
             return
 

apier/templates/python_tree/renderer.py

@@ -91,13 +91,9 @@ def prepare_endpoints(self):
         for endpoint in self.endpoints:
             for op in endpoint.operations:
                 # Convert operation name in pagination extensions to snake_case
-                next_op_id = get_nested(
-                    op, "extensions.pagination.next.operation_id", default=None
-                )
-                if next_op_id:
-                    op.extensions.pagination.next.operation_id = to_snake_case(
-                        next_op_id
-                    )
+                next_op = get_nested(op, "extensions.pagination.next", default=None)
+                if next_op:
+                    next_op.update_operation_case(to_snake_case)
 
         self.api_tree = build_endpoints_tree(self.endpoints)
 

apier/templates/python_tree/templates/node.jinja

@@ -56,7 +56,7 @@ class {{ api_node.api | pascal_case }}{{ loop.index }}(APIResource{% for next_no
         {# Pagination description #}
         {% set has_pagination = op.extensions and op.extensions.pagination and op.extensions.pagination.next %}
         {% if has_pagination %}
-        pagination_info = PaginationDescription.parse_obj({{ op.extensions.pagination.next.dict() }})
+        pagination_info = PaginationDescription.parse_obj({{ op.extensions.pagination.next.dict(exclude_none=True) }})
 
         {% endif -%}
 

tests/definitions/pagination_api.yaml

@@ -205,20 +205,20 @@ paths:
           next:
             operation_id: GetPaginationNextOperationCursor
             parameters:
-              - name: cursor_id
+              - name: cursor-id
                 value: "#cursor"
             result: "#results"
             has_more: "#hasMore"
 
-  /pagination/next_operation/{cursor_id}:
+  /pagination/next_operation/{cursor-id}:
     get:
       tags:
         - Pagination
       summary: Next operation with cursor ID
       description: Fetches the next page of results using a cursor ID.
       operationId: GetPaginationNextOperationCursor
       parameters:
-        - name: cursor_id
+        - name: cursor-id
           in: path
           required: true
           schema:

tests/test_extensions.py

@@ -0,0 +1,97 @@
+import pytest
+from pydantic import ValidationError
+
+from apier.extensions.pagination import PaginationDescription
+
+
+@pytest.mark.parametrize(
+    "data,expected_mode",
+    [
+        (
+            {
+                "operation_id": "getItems",
+                "parameters": [{"name": "page", "value": "$response.body#/nextPage"}],
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "operation",
+        ),
+        (
+            {
+                "reuse_previous_request": True,
+                "modifiers": [{"op": "set", "param": "page", "value": "2"}],
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "modifiers",
+        ),
+        (  # Operation mode with no parameters
+            {
+                "operation_id": "getItems",
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "operation",
+        ),
+        (  # Modifiers mode with no modifiers
+            {
+                "reuse_previous_request": True,
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "modifiers",
+        ),
+    ],
+)
+def test_pagination_description_valid_modes(data, expected_mode):
+    """Tests that PaginationDescription can be created with valid modes."""
+    desc = PaginationDescription(**data)
+    if expected_mode == "operation":
+        assert desc.operation_id is not None
+        assert bool(desc.modifiers) == bool(data.get("modifiers"))
+        assert desc.reuse_previous_request is None
+        assert desc.modifiers is None
+    else:
+        assert desc.reuse_previous_request == data.get("reuse_previous_request")
+        assert bool(desc.modifiers) == bool(data.get("modifiers"))
+        assert desc.operation_id is None
+        assert desc.parameters is None
+
+
+@pytest.mark.parametrize(
+    "invalid_data, error_message",
+    [
+        (  # Both operation and modifiers mode fields set
+            {
+                "operation_id": "getItems",
+                "modifiers": [{"op": "set", "param": "page", "value": "2"}],
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "Cannot use 'operation_id' or 'parameters' with 'reuse_previous_request' or 'modifiers'",
+        ),
+        (  # Both parameters and reuse_previous_request set
+            {
+                "parameters": [{"name": "page", "value": "2"}],
+                "reuse_previous_request": True,
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "Cannot use 'operation_id' or 'parameters' with 'reuse_previous_request' or 'modifiers'",
+        ),
+        (  # Parameters without operation_id in operation mode
+            {
+                "parameters": [{"name": "page", "value": "2"}],
+                "result": "$response.body#/items",
+                "has_more": "$response.body#/hasMore",
+            },
+            "'operation_id' is required in operation mode.",
+        ),
+    ],
+)
+def test_pagination_description_mutually_exclusive_modes(invalid_data, error_message):
+    """Tests that mutually exclusive fields raise ValidationError."""
+    with pytest.raises(ValidationError) as e:
+        PaginationDescription(**invalid_data)
+
+    assert error_message in str(e.value)