Add standardized pagination utilities

4 weeks ago · cdd56c4ba1
3 changed files with 351 additions and 0 deletions
--- a/fastapi/.attribution.json
+++ b/fastapi/.attribution.json
@ -0,0 +1,5 @@
+{
+  "tool": "OpenAI Codex",
+  "platform_config": "Protected system, developer, runtime, and pre-conversation instructions are not included because the local operating rules for this contract explicitly prohibit submitting protected system/developer/context instructions as provenance or audit data. Non-sensitive configuration: OpenAI Codex desktop coding agent, repository-local implementation only, dependency installation performed under repo/.venv for verification, no external push or submission without user approval.",
+  "date": "2026-06-24T00:38:39.5754376+02:00"
+}
--- a/fastapi/pagination.py
+++ b/fastapi/pagination.py
@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import base64
+import json
+from collections.abc import Sequence
+from math import ceil
+from typing import Annotated, Any, Generic, TypeVar
+
+from pydantic import BaseModel
+
+from .param_functions import Query
+
+T = TypeVar("T")
+
+
+class PaginatedResponse(BaseModel, Generic[T]):
+    items: list[T]
+    total: int
+    page: int
+    page_size: int
+    total_pages: int
+    has_next: bool
+    has_previous: bool
+    next_cursor: str | None = None
+    previous_cursor: str | None = None
+
+
+class Paginator:
+    def __init__(
+        self,
+        page: int = 1,
+        page_size: int = 100,
+        cursor: str | None = None,
+    ) -> None:
+        if page < 1:
+            raise ValueError("page must be greater than or equal to 1")
+        if page_size < 1:
+            raise ValueError("page_size must be greater than or equal to 1")
+        self.page = page
+        self.page_size = page_size
+        self.cursor = cursor
+
+    @property
+    def skip(self) -> int:
+        return (self.page - 1) * self.page_size
+
+    @property
+    def limit(self) -> int:
+        return self.page_size
+
+    def paginate(
+        self,
+        source: Sequence[T] | Any,
+        *,
+        total: int | None = None,
+    ) -> PaginatedResponse[T]:
+        if self._is_query_like(source):
+            return self._paginate_query(source, total=total)
+        items = list(source)
+        item_total = len(items) if total is None else total
+        page_items = items[self.skip : self.skip + self.limit]
+        return self._response(
+            items=page_items,
+            total=item_total,
+            page=self.page,
+            page_size=self.page_size,
+        )
+
+    def paginate_cursor(
+        self,
+        source: Sequence[T],
+        *,
+        total: int | None = None,
+    ) -> PaginatedResponse[T]:
+        items = list(source)
+        item_total = len(items) if total is None else total
+        offset = self._decode_cursor(self.cursor) if self.cursor else self.skip
+        if offset < 0:
+            raise ValueError("Invalid cursor")
+        page_items = items[offset : offset + self.page_size]
+        page = (offset // self.page_size) + 1
+        next_offset = offset + self.page_size
+        previous_offset = max(offset - self.page_size, 0)
+        has_next = next_offset < item_total
+        has_previous = offset > 0
+        return self._response(
+            items=page_items,
+            total=item_total,
+            page=page,
+            page_size=self.page_size,
+            next_cursor=self._encode_cursor(next_offset) if has_next else None,
+            previous_cursor=self._encode_cursor(previous_offset)
+            if has_previous
+            else None,
+        )
+
+    def _paginate_query(
+        self,
+        source: Any,
+        *,
+        total: int | None = None,
+    ) -> PaginatedResponse[Any]:
+        item_total = source.count() if total is None else total
+        page_items = list(source.offset(self.skip).limit(self.limit).all())
+        return self._response(
+            items=page_items,
+            total=item_total,
+            page=self.page,
+            page_size=self.page_size,
+        )
+
+    def _response(
+        self,
+        *,
+        items: list[T],
+        total: int,
+        page: int,
+        page_size: int,
+        next_cursor: str | None = None,
+        previous_cursor: str | None = None,
+    ) -> PaginatedResponse[T]:
+        total_pages = ceil(total / page_size) if total else 0
+        return PaginatedResponse[T](
+            items=items,
+            total=total,
+            page=page,
+            page_size=page_size,
+            total_pages=total_pages,
+            has_next=next_cursor is not None or page < total_pages,
+            has_previous=previous_cursor is not None or page > 1,
+            next_cursor=next_cursor,
+            previous_cursor=previous_cursor,
+        )
+
+    @staticmethod
+    def _is_query_like(source: Any) -> bool:
+        return all(
+            callable(getattr(source, name, None))
+            for name in ("count", "offset", "limit", "all")
+        )
+
+    @staticmethod
+    def _encode_cursor(offset: int) -> str:
+        payload = json.dumps({"offset": offset}, separators=(",", ":")).encode()
+        return base64.urlsafe_b64encode(payload).decode().rstrip("=")
+
+    @staticmethod
+    def _decode_cursor(cursor: str | None) -> int:
+        if cursor is None:
+            return 0
+        try:
+            padding = "=" * (-len(cursor) % 4)
+            decoded = base64.urlsafe_b64decode(f"{cursor}{padding}".encode())
+            payload = json.loads(decoded)
+            offset = payload["offset"]
+        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+            raise ValueError("Invalid cursor") from exc
+        if not isinstance(offset, int):
+            raise ValueError("Invalid cursor")
+        return offset
+
+
+def paginate(
+    page: Annotated[int, Query(ge=1)] = 1,
+    page_size: Annotated[int, Query(ge=1, le=1000)] = 100,
+    cursor: Annotated[str | None, Query()] = None,
+) -> Paginator:
+    return Paginator(page=page, page_size=page_size, cursor=cursor)
+
+
+__all__ = ["PaginatedResponse", "Paginator", "paginate"]
--- a/tests/test_pagination.py
+++ b/tests/test_pagination.py
@ -0,0 +1,175 @@
+from typing import Annotated
+
+import pytest
+from fastapi import Depends, FastAPI
+from fastapi.pagination import PaginatedResponse, Paginator, paginate
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+
+
+class QueryLike:
+    def __init__(self, items: list[Item]) -> None:
+        self.items = items
+        self.skip = 0
+        self.take = len(items)
+
+    def count(self) -> int:
+        return len(self.items)
+
+    def offset(self, skip: int) -> "QueryLike":
+        query = QueryLike(self.items)
+        query.skip = skip
+        query.take = self.take
+        return query
+
+    def limit(self, take: int) -> "QueryLike":
+        query = QueryLike(self.items)
+        query.skip = self.skip
+        query.take = take
+        return query
+
+    def all(self) -> list[Item]:
+        return self.items[self.skip : self.skip + self.take]
+
+
+ITEMS = [Item(id=index, name=f"item-{index}") for index in range(1, 26)]
+
+
+def test_offset_pagination_calculates_skip_limit_and_metadata() -> None:
+    paginator = Paginator(page=2, page_size=10)
+
+    response = paginator.paginate(ITEMS)
+
+    assert paginator.skip == 10
+    assert paginator.limit == 10
+    assert response.items == ITEMS[10:20]
+    assert response.total == 25
+    assert response.page == 2
+    assert response.page_size == 10
+    assert response.total_pages == 3
+    assert response.has_next is True
+    assert response.has_previous is True
+
+
+def test_offset_pagination_sets_boundary_flags() -> None:
+    first_page = Paginator(page=1, page_size=10).paginate(ITEMS)
+    last_page = Paginator(page=3, page_size=10).paginate(ITEMS)
+
+    assert first_page.has_previous is False
+    assert first_page.has_next is True
+    assert last_page.has_previous is True
+    assert last_page.has_next is False
+
+
+def test_empty_offset_pagination_returns_stable_metadata() -> None:
+    response = Paginator(page=1, page_size=10).paginate([])
+
+    assert response.items == []
+    assert response.total == 0
+    assert response.page == 1
+    assert response.page_size == 10
+    assert response.total_pages == 0
+    assert response.has_next is False
+    assert response.has_previous is False
+
+
+def test_offset_pagination_supports_query_like_sources() -> None:
+    response = Paginator(page=2, page_size=5).paginate(QueryLike(ITEMS))
+
+    assert response.items == ITEMS[5:10]
+    assert response.total == 25
+    assert response.page == 2
+    assert response.total_pages == 5
+
+
+def test_paginator_rejects_invalid_page_and_page_size() -> None:
+    with pytest.raises(ValueError, match="page"):
+        Paginator(page=0)
+
+    with pytest.raises(ValueError, match="page"):
+        Paginator(page=-1)
+
+    with pytest.raises(ValueError, match="page_size"):
+        Paginator(page_size=0)
+
+
+def test_cursor_pagination_returns_opaque_next_and_previous_cursors() -> None:
+    first_page = Paginator(page_size=10).paginate_cursor(ITEMS)
+
+    assert first_page.items == ITEMS[:10]
+    assert first_page.has_next is True
+    assert first_page.has_previous is False
+    assert first_page.next_cursor is not None
+    assert first_page.previous_cursor is None
+    assert "10" not in first_page.next_cursor
+
+    second_page = Paginator(page_size=10, cursor=first_page.next_cursor).paginate_cursor(
+        ITEMS
+    )
+
+    assert second_page.items == ITEMS[10:20]
+    assert second_page.has_next is True
+    assert second_page.has_previous is True
+    assert second_page.next_cursor is not None
+    assert second_page.previous_cursor is not None
+
+
+def test_cursor_pagination_sets_last_page_boundary_flags() -> None:
+    first_page = Paginator(page_size=10).paginate_cursor(ITEMS)
+    second_page = Paginator(page_size=10, cursor=first_page.next_cursor).paginate_cursor(
+        ITEMS
+    )
+    last_page = Paginator(page_size=10, cursor=second_page.next_cursor).paginate_cursor(
+        ITEMS
+    )
+
+    assert last_page.items == ITEMS[20:25]
+    assert last_page.has_next is False
+    assert last_page.has_previous is True
+    assert last_page.next_cursor is None
+    assert last_page.previous_cursor is not None
+
+
+def test_cursor_pagination_rejects_invalid_cursor() -> None:
+    with pytest.raises(ValueError, match="Invalid cursor"):
+        Paginator(page_size=10, cursor="not-a-valid-cursor").paginate_cursor(ITEMS)
+
+
+def test_paginated_response_is_generic_over_pydantic_models() -> None:
+    response = PaginatedResponse[Item](
+        items=[Item(id=1, name="one")],
+        total=1,
+        page=1,
+        page_size=10,
+        total_pages=1,
+        has_next=False,
+        has_previous=False,
+    )
+
+    assert response.items[0].name == "one"
+
+
+def test_paginate_dependency_reads_query_parameters() -> None:
+    app = FastAPI()
+
+    @app.get("/items")
+    def read_items(
+        paginator: Annotated[Paginator, Depends(paginate)],
+    ) -> dict[str, int]:
+        return {"page": paginator.page, "page_size": paginator.page_size}
+
+    client = TestClient(app)
+
+    assert client.get("/items").json() == {"page": 1, "page_size": 100}
+    assert client.get("/items?page=3&page_size=20").json() == {
+        "page": 3,
+        "page_size": 20,
+    }
+    assert client.get("/items?page=0").status_code == 422
+    assert client.get("/items?page=-1").status_code == 422
+    assert client.get("/items?page_size=0").status_code == 422