diff --git a/README.md b/README.md index d7639da..2e8ed56 100644 --- a/README.md +++ b/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) diff --git a/src/kernel/pagination.py b/src/kernel/pagination.py index cdf83c2..6e3ebb5 100644 --- a/src/kernel/pagination.py +++ b/src/kernel/pagination.py @@ -39,13 +39,17 @@ def has_next_page(self) -> bool: @override 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 next_offset is None: # type: ignore[unreachable] + if self.has_more: + raise RuntimeError( + "Server reported X-Has-More: true without an X-Next-Offset header; " + "refusing to silently truncate pagination" + ) + return None + + # X-Next-Offset already holds the offset where the next page starts; + # adding the current page length on top skips a full page per iteration. + return PageInfo(params={"offset": next_offset}) @classmethod def build(cls: Type[_BaseModelT], *, response: Response, data: object) -> _BaseModelT: # noqa: ARG003 @@ -82,13 +86,17 @@ def has_next_page(self) -> bool: @override 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 next_offset is None: # type: ignore[unreachable] + if self.has_more: + raise RuntimeError( + "Server reported X-Has-More: true without an X-Next-Offset header; " + "refusing to silently truncate pagination" + ) + return None + + # X-Next-Offset already holds the offset where the next page starts; + # adding the current page length on top skips a full page per iteration. + return PageInfo(params={"offset": next_offset}) @classmethod def build(cls: Type[_BaseModelT], *, response: Response, data: object) -> _BaseModelT: # noqa: ARG003 diff --git a/tests/test_pagination.py b/tests/test_pagination.py new file mode 100644 index 0000000..46723c9 --- /dev/null +++ b/tests/test_pagination.py @@ -0,0 +1,62 @@ +from typing import Any, List, Type, Union, Optional + +import httpx +import pytest + +from kernel.pagination import AsyncOffsetPagination, SyncOffsetPagination + +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: + # Covers the 0 sentinel the API emits on last pages: has_more is false + # whenever next_offset is 0, and has_more gates first. + page = _page(cls, items=[{}] * 50, next_offset=0, has_more=False) + assert page.has_next_page() is False + + +@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()