diff --git a/fastapi/.attribution.json b/fastapi/.attribution.json new file mode 100644 index 000000000..f26c0574e --- /dev/null +++ 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" +} diff --git a/fastapi/pagination.py b/fastapi/pagination.py new file mode 100644 index 000000000..f47453785 --- /dev/null +++ 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"] diff --git a/tests/test_pagination.py b/tests/test_pagination.py new file mode 100644 index 000000000..5abf966b1 --- /dev/null +++ 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