add paginator test cases and support for json next link paginator

openapi_python_client/parser/const.py

@@ -13,6 +13,7 @@
 RE_LIMIT_PARAM = re.compile(r"(?i)(limit|per_page|page_size|size)", re.IGNORECASE)
 RE_TOTAL_PROPERTY = re.compile(r"(?i)(total|count)", re.IGNORECASE)
 RE_CURSOR_PARAM = re.compile(r"(?i)(cursor|after|since)", re.IGNORECASE)
+RE_NEXT_PROPERTY = re.compile(r"(?i)(next|next_url|more)", re.IGNORECASE)
 
 
 # content path discovery

openapi_python_client/parser/endpoints.py

@@ -207,7 +207,7 @@ def keyword_arguments(self) -> List[Parameter]:
         return list(ret)
 
     @property
-    def pagination_args(self) -> Optional[Dict[str, str]]:
+    def pagination_args(self) -> Optional[Dict[str, Union[str, int]]]:
         return self.pagination.paginator_config if self.pagination else None
 
     def all_arguments(self) -> List[Parameter]:

openapi_python_client/parser/pagination.py

@@ -1,18 +1,18 @@
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, Any
+from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, Union
 
 from openapi_python_client.parser.models import DataPropertyPath
 
 if TYPE_CHECKING:
     from openapi_python_client.parser.endpoints import Endpoint, Parameter
 
-from .const import RE_OFFSET_PARAM, RE_LIMIT_PARAM, RE_TOTAL_PROPERTY, RE_CURSOR_PARAM
+from .const import RE_OFFSET_PARAM, RE_LIMIT_PARAM, RE_TOTAL_PROPERTY, RE_CURSOR_PARAM, RE_NEXT_PROPERTY
 
 
 @dataclass
 class Pagination:
     pagination_params: List["Parameter"] = field(default_factory=list)
-    paginator_config: Dict[str, str] = None
+    paginator_config: Dict[str, Union[str, int]] = None
 
     @property
     def param_names(self) -> List[str]:
@@ -38,28 +38,31 @@ def from_endpoint(cls, endpoint: "Endpoint") -> Optional["Pagination"]:
             if RE_CURSOR_PARAM.match(param_name):
                 cursor_params.append(param)
 
+        #
+        # Detect cursor
+        #
         cursor_props: List[Tuple["Parameter", DataPropertyPath]] = []
         for cursor_param in cursor_params:
             # Try to response property to feed into the cursor param
-            prop = cursor_param.find_input_property(resp.content_schema, fallback=None)
-            if prop:
+            if prop := cursor_param.find_input_property(resp.content_schema, fallback=None):
                 cursor_props.append((cursor_param, prop))
 
-        pagination_config: Optional[Dict[str, Any]] = None
         # Prefer the least nested cursor prop
         if cursor_props:
             cursor_props.sort(key=lambda x: len(x[1].path))
             cursor_param, cursor_prop = cursor_props[0]
-            pagination_config = {
-                "type": "cursor",
-                "cursor_path": cursor_prop.json_path,
-                "cursor_param": cursor_param.name,
-            }
             return cls(
-                paginator_config=pagination_config,
+                paginator_config={
+                    "type": "cursor",
+                    "cursor_path": cursor_prop.json_path,
+                    "cursor_param": cursor_param.name,
+                },
                 pagination_params=[cursor_param],
             )
 
+        #
+        # Detect offset - limit
+        #
         offset_props: List[Tuple["Parameter", DataPropertyPath]] = []
         offset_param: Optional["Parameter"] = None
         limit_param: Optional["Parameter"] = None
@@ -82,17 +85,28 @@ def from_endpoint(cls, endpoint: "Endpoint") -> Optional["Pagination"]:
         total_prop = resp.content_schema.crawled_properties.find_property(RE_TOTAL_PROPERTY, require_type="integer")
 
         if offset_param and limit_param and limit_initial and total_prop:
-            pagination_config = {
-                "type": "offset",
-                "initial_limit": limit_initial,
-                "offset_param": offset_param.name,
-                "limit_param": limit_param.name,
-                "total_path": total_prop.json_path,
-            }
             return cls(
-                paginator_config=pagination_config,
+                paginator_config={
+                    "type": "offset",
+                    "initial_limit": limit_initial,
+                    "offset_param": offset_param.name,
+                    "limit_param": limit_param.name,
+                    "total_path": total_prop.json_path,
+                },
+                pagination_params=[offset_param, limit_param],
+            )
+
+        #
+        # Detect json_links
+        #
+        next_prop = resp.content_schema.crawled_properties.find_property(RE_NEXT_PROPERTY, require_type="string")
+        if next_prop:
+            return cls(
+                paginator_config={"type": "json_links", "next_url_path": next_prop.json_path},
                 pagination_params=[offset_param, limit_param],
             )
 
-        # No pagination detected
+        #
+        # Nothing found :(
+        #
         return None

tests/cases/test_specs/pagination_specs.yml

@@ -0,0 +1,93 @@
+openapi: 3.0.0
+info:
+  title: 'pagination'
+  version: 1.0.0
+  description: 'different pagination examples'
+servers:
+- url: 'https://pokeapi.co/'
+
+paths:
+  /offset_limit_pagination_1/:
+    get:
+      operationId: offset_limit_pagination_1
+      parameters:
+      - in: query
+        name: limit
+        schema:
+          description: Number of results to return per page.
+          title: Limit
+          type: integer
+      - in: query
+        name: offset
+        schema:
+          description: The initial index from which to return the results.
+          title: Offset
+          type: integer
+      responses:
+        '200':
+          description: "OK"
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  count:
+                    type: integer
+                    example: 3
+                  next:
+                    type: string
+                    nullable: true
+                    example: https://pokeapi.co/api/v2/pokemon/?offset=20&limit=20
+                  previous:
+                    type: string
+                    nullable: true
+                  results:
+                    type: array
+
+  /cursor_pagination_1/:
+    get:
+      operationId: cursor_pagination_1
+      parameters:
+      - in: query
+        name: cursor
+        schema:
+          description: Put cursor here
+          title: Cursor
+          type: string
+      responses:
+        '200':
+          description: "OK"
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  cursor:
+                    type: string
+                  results:
+                    type: array
+
+
+  /json_links_pagination_1/:
+    get:
+      operationId: json_links_pagination_1
+      responses:
+        '200':
+          description: "OK"
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  count:
+                    type: integer
+                    example: 3
+                  next:
+                    type: string
+                    nullable: true
+                    example: https://pokeapi.co/api/v2/pokemon/?offset=20&limit=20
+                  previous:
+                    type: string
+                    nullable: true
+                  results:
+                    type: array

tests/e2e/test_paginator_cases.py

@@ -0,0 +1,40 @@
+from typing import Dict, Any
+
+import pytest
+
+from tests.e2e.utils import get_source_from_open_api, get_dict_from_open_api
+from tests.cases import get_test_case_path
+
+
+@pytest.fixture(scope="module")
+def paginators() -> Dict[str, Any]:
+    case_path = get_test_case_path("pagination_specs.yml")
+    # validate that source will work
+    get_source_from_open_api(case_path)
+
+    # get dict and save paginator info into dict
+    rendered_dict = get_dict_from_open_api(case_path)
+    return {
+        entry["name"]: entry.get("endpoint").get("paginator") for entry in rendered_dict["resources"]  # type: ignore
+    }
+
+
+def test_offset_limit_pagination_1(paginators: Dict[str, Any]) -> None:
+    assert paginators["offset_limit_pagination_1"] == {
+        "initial_limit": 20,
+        "limit_param": "limit",
+        "offset_param": "offset",
+        "type": "offset",
+        "total_path": "count",
+    }
+
+
+def test_json_links_pagination_1(paginators: Dict[str, Any]) -> None:
+    assert paginators["json_links_pagination_1"] == {
+        "next_url_path": "next",
+        "type": "json_links",
+    }
+
+
+def test_json_cursor_1(paginators: Dict[str, Any]) -> None:
+    assert paginators["cursor_pagination_1"] == {"cursor_param": "cursor", "cursor_path": "cursor", "type": "cursor"}