Merge pull request #119 from kernel/fix-offset-pagination-page-skip

Fix offset auto-pagination skipping every other page

Author: IlyaasK

README.md

@@ -185,8 +185,8 @@ first_page = await client.deployments.list(
 )
 
 print(
-    f"the current start offset for this page: {first_page.next_offset}"
-)  # => "the current start offset for this page: 1"
+    f"the offset where the next page starts: {first_page.next_offset}"
+)  # => "the offset where the next page starts: 2"
 for deployment in first_page.items:
     print(deployment.id)
 

src/kernel/pagination.py

@@ -40,12 +40,20 @@ def has_next_page(self) -> bool:
     def next_page_info(self) -> Optional[PageInfo]:
         next_offset = self.next_offset
         if next_offset is None:
-            return None  # type: ignore[unreachable]
+            if self.has_more:  # type: ignore[unreachable]
+                raise RuntimeError(
+                    "Server reported X-Has-More: true without an X-Next-Offset header; "
+                    "refusing to silently truncate pagination"
+                )
+            return None
 
-        length = len(self._get_page_items())
-        current_count = next_offset + length
+        # X-Next-Offset is the next page's absolute start, or 0 on the last page
+        # (the API's stop sentinel). Only a positive offset advances; the old code
+        # added the current page length on top, skipping a page per iteration.
+        if next_offset == 0:
+            return None
 
-        return PageInfo(params={"offset": current_count})
+        return PageInfo(params={"offset": next_offset})
 
     @classmethod
     def build(cls: Type[_BaseModelT], *, response: Response, data: object) -> _BaseModelT:  # noqa: ARG003
@@ -83,12 +91,20 @@ def has_next_page(self) -> bool:
     def next_page_info(self) -> Optional[PageInfo]:
         next_offset = self.next_offset
         if next_offset is None:
-            return None  # type: ignore[unreachable]
-
-        length = len(self._get_page_items())
-        current_count = next_offset + length
-
-        return PageInfo(params={"offset": current_count})
+            if self.has_more:  # type: ignore[unreachable]
+                raise RuntimeError(
+                    "Server reported X-Has-More: true without an X-Next-Offset header; "
+                    "refusing to silently truncate pagination"
+                )
+            return None
+
+        # X-Next-Offset is the next page's absolute start, or 0 on the last page
+        # (the API's stop sentinel). Only a positive offset advances; the old code
+        # added the current page length on top, skipping a page per iteration.
+        if next_offset == 0:
+            return None
+
+        return PageInfo(params={"offset": next_offset})
 
     @classmethod
     def build(cls: Type[_BaseModelT], *, response: Response, data: object) -> _BaseModelT:  # noqa: ARG003

tests/test_pagination.py

@@ -0,0 +1,66 @@
+from typing import Any, List, Type, Union, Optional
+
+import httpx
+import pytest
+
+from kernel.pagination import SyncOffsetPagination, AsyncOffsetPagination
+
+PageClass = Union[Type[SyncOffsetPagination[Any]], Type[AsyncOffsetPagination[Any]]]
+
+# build() and next_page_info() are plain synchronous methods on both classes,
+# so both generated variants are pinned against drifting apart on regeneration.
+both_classes = pytest.mark.parametrize("cls", [SyncOffsetPagination, AsyncOffsetPagination])
+
+
+def _page(cls: PageClass, *, items: List[Any], next_offset: Optional[int], has_more: Optional[bool]) -> Any:
+    headers: dict[str, str] = {}
+    if next_offset is not None:
+        headers["X-Next-Offset"] = str(next_offset)
+    if has_more is not None:
+        headers["X-Has-More"] = "true" if has_more else "false"
+    response = httpx.Response(200, headers=headers)
+    return cls.build(response=response, data=items)
+
+
+@both_classes
+def test_next_page_starts_at_exactly_x_next_offset(cls: PageClass) -> None:
+    # X-Next-Offset already holds the next page's start. Adding the current
+    # page length on top (the old behavior) skipped a full page per iteration.
+    page = _page(cls, items=[{}] * 100, next_offset=100, has_more=True)
+    info = page.next_page_info()
+    assert info is not None
+    assert info.params == {"offset": 100}
+
+
+@both_classes
+def test_stops_cleanly_when_last_page_omits_x_next_offset(cls: PageClass) -> None:
+    page = _page(cls, items=[{}] * 50, next_offset=None, has_more=False)
+    assert page.next_page_info() is None
+    assert page.has_next_page() is False
+
+
+@both_classes
+def test_stops_when_x_has_more_false(cls: PageClass) -> None:
+    page = _page(cls, items=[{}] * 50, next_offset=200, has_more=False)
+    assert page.has_next_page() is False
+
+
+@both_classes
+def test_stops_on_next_offset_zero_sentinel(cls: PageClass) -> None:
+    # 0 is the API's last-page sentinel. Assert via next_page_info directly,
+    # with has_more=True, so the has_more gate cannot mask the 0 handling.
+    page = _page(cls, items=[{}] * 50, next_offset=0, has_more=True)
+    assert page.next_page_info() is None
+
+
+@both_classes
+def test_stops_on_empty_page(cls: PageClass) -> None:
+    page = _page(cls, items=[], next_offset=300, has_more=True)
+    assert page.has_next_page() is False
+
+
+@both_classes
+def test_refuses_to_silently_truncate_on_contradictory_headers(cls: PageClass) -> None:
+    page = _page(cls, items=[{}] * 100, next_offset=None, has_more=True)
+    with pytest.raises(RuntimeError, match="refusing to silently truncate"):
+        page.has_next_page()