feat: unified paginate() endpoint with typed pagination responses (#134)

* feat: unified paginate() endpoint with typed pagination responses * docs: unified paginate() endpoint * fix: add tests
2026-04-15 22:26:25 +02:00 · 2026-03-14 20:23:20 +01:00
parent 81407c3038
commit 19c013bdec
11 changed files with 811 additions and 86 deletions
--- a/tests/test_crud.py
+++ b/tests/test_crud.py
@@ -6,7 +6,7 @@ import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload

-from fastapi_toolsets.crud import CrudFactory
+from fastapi_toolsets.crud import CrudFactory, PaginationType
 from fastapi_toolsets.crud.factory import AsyncCrud
 from fastapi_toolsets.exceptions import NotFoundError

@@ -2452,3 +2452,72 @@ class TestCursorPaginateColumnTypes:
        page1_ids = {p.id for p in page1.data}
        page2_ids = {p.id for p in page2.data}
        assert page1_ids.isdisjoint(page2_ids)
+
+
+class TestPaginate:
+    """Tests for the unified paginate() method."""
+
+    @pytest.mark.anyio
+    async def test_offset_pagination(self, db_session: AsyncSession):
+        """paginate() with OFFSET returns OffsetPaginatedResponse."""
+        from fastapi_toolsets.schemas import OffsetPagination
+
+        await RoleCrud.create(db_session, RoleCreate(name="admin"))
+        await RoleCrud.create(db_session, RoleCreate(name="user"))
+
+        result = await RoleCrud.paginate(
+            db_session,
+            pagination_type=PaginationType.OFFSET,
+            schema=RoleRead,
+        )
+
+        assert isinstance(result.pagination, OffsetPagination)
+        assert result.pagination_type == PaginationType.OFFSET
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination(self, db_session: AsyncSession):
+        """paginate() with CURSOR returns CursorPaginatedResponse."""
+        from fastapi_toolsets.schemas import CursorPagination
+
+        await RoleCursorCrud.create(db_session, RoleCreate(name="admin"))
+
+        result = await RoleCursorCrud.paginate(
+            db_session,
+            pagination_type=PaginationType.CURSOR,
+            schema=RoleRead,
+        )
+
+        assert isinstance(result.pagination, CursorPagination)
+        assert result.pagination_type == PaginationType.CURSOR
+
+    @pytest.mark.anyio
+    async def test_invalid_items_per_page_raises(self, db_session: AsyncSession):
+        """paginate() raises ValueError when items_per_page < 1."""
+        with pytest.raises(ValueError, match="items_per_page"):
+            await RoleCrud.paginate(
+                db_session,
+                pagination_type=PaginationType.OFFSET,
+                items_per_page=0,
+                schema=RoleRead,
+            )
+
+    @pytest.mark.anyio
+    async def test_invalid_page_raises(self, db_session: AsyncSession):
+        """paginate() raises ValueError when page < 1 for offset pagination."""
+        with pytest.raises(ValueError, match="page"):
+            await RoleCrud.paginate(
+                db_session,
+                pagination_type=PaginationType.OFFSET,
+                page=0,
+                schema=RoleRead,
+            )
+
+    @pytest.mark.anyio
+    async def test_unknown_pagination_type_raises(self, db_session: AsyncSession):
+        """paginate() raises ValueError for unknown pagination_type."""
+        with pytest.raises(ValueError, match="Unknown pagination_type"):
+            await RoleCrud.paginate(
+                db_session,
+                pagination_type="unknown",
+                schema=RoleRead,
+            )  # type: ignore[no-matching-overload]
--- a/tests/test_example_pagination_search.py
+++ b/tests/test_example_pagination_search.py
@@ -393,3 +393,105 @@ class TestCursorSorting:
        body = resp.json()
        assert body["error_code"] == "SORT-422"
        assert body["status"] == "FAIL"
+
+
+class TestPaginateUnified:
+    """Tests for the unified GET /articles/ endpoint using paginate()."""
+
+    @pytest.mark.anyio
+    async def test_defaults_to_offset_pagination(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """Without pagination_type, defaults to offset pagination."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "offset"
+        assert "total_count" in body["pagination"]
+        assert body["pagination"]["total_count"] == 3
+
+    @pytest.mark.anyio
+    async def test_explicit_offset_pagination(self, client: AsyncClient, ex_db_session):
+        """pagination_type=offset returns OffsetPagination metadata."""
+        await seed(ex_db_session)
+
+        resp = await client.get(
+            "/articles/?pagination_type=offset&page=1&items_per_page=2"
+        )
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "offset"
+        assert body["pagination"]["total_count"] == 3
+        assert body["pagination"]["page"] == 1
+        assert body["pagination"]["has_more"] is True
+        assert len(body["data"]) == 2
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination_type(self, client: AsyncClient, ex_db_session):
+        """pagination_type=cursor returns CursorPagination metadata."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/?pagination_type=cursor&items_per_page=2")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "cursor"
+        assert "next_cursor" in body["pagination"]
+        assert "total_count" not in body["pagination"]
+        assert body["pagination"]["has_more"] is True
+        assert len(body["data"]) == 2
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination_navigate_pages(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """Cursor from first page can be used to fetch the next page."""
+        await seed(ex_db_session)
+
+        first = await client.get("/articles/?pagination_type=cursor&items_per_page=2")
+        assert first.status_code == 200
+        first_body = first.json()
+        next_cursor = first_body["pagination"]["next_cursor"]
+        assert next_cursor is not None
+
+        second = await client.get(
+            f"/articles/?pagination_type=cursor&items_per_page=2&cursor={next_cursor}"
+        )
+        assert second.status_code == 200
+        second_body = second.json()
+        assert second_body["pagination_type"] == "cursor"
+        assert second_body["pagination"]["has_more"] is False
+        assert len(second_body["data"]) == 1
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination_with_search(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """paginate() with cursor type respects search parameter."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/?pagination_type=cursor&search=fastapi")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "cursor"
+        assert len(body["data"]) == 1
+        assert body["data"][0]["title"] == "FastAPI tips"
+
+    @pytest.mark.anyio
+    async def test_offset_pagination_with_filter(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """paginate() with offset type respects filter_by parameter."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/?pagination_type=offset&status=published")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "offset"
+        assert body["pagination"]["total_count"] == 2
--- a/tests/test_schemas.py
+++ b/tests/test_schemas.py
@@ -6,9 +6,12 @@ from pydantic import ValidationError
 from fastapi_toolsets.schemas import (
    ApiError,
    CursorPagination,
+    CursorPaginatedResponse,
    ErrorResponse,
    OffsetPagination,
+    OffsetPaginatedResponse,
    PaginatedResponse,
+    PaginationType,
    Response,
    ResponseStatus,
 )
@@ -312,11 +315,6 @@ class TestPaginatedResponse:

    def test_generic_type_hint(self):
        """PaginatedResponse supports generic type hints."""
-
-        class UserOut:
-            id: int
-            name: str
-
        pagination = OffsetPagination(
            total_count=1,
            items_per_page=10,
@@ -371,6 +369,191 @@ class TestPaginatedResponse:
        assert isinstance(response.pagination, CursorPagination)


+class TestPaginationType:
+    """Tests for PaginationType enum."""
+
+    def test_offset_value(self):
+        """OFFSET has string value 'offset'."""
+        assert PaginationType.OFFSET == "offset"
+        assert PaginationType.OFFSET.value == "offset"
+
+    def test_cursor_value(self):
+        """CURSOR has string value 'cursor'."""
+        assert PaginationType.CURSOR == "cursor"
+        assert PaginationType.CURSOR.value == "cursor"
+
+    def test_is_string_enum(self):
+        """PaginationType is a string enum."""
+        assert isinstance(PaginationType.OFFSET, str)
+        assert isinstance(PaginationType.CURSOR, str)
+
+    def test_members(self):
+        """PaginationType has exactly two members."""
+        assert set(PaginationType) == {PaginationType.OFFSET, PaginationType.CURSOR}
+
+
+class TestOffsetPaginatedResponse:
+    """Tests for OffsetPaginatedResponse schema."""
+
+    def test_pagination_type_is_offset(self):
+        """pagination_type is always PaginationType.OFFSET."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert response.pagination_type is PaginationType.OFFSET
+
+    def test_pagination_type_serializes_to_string(self):
+        """pagination_type serializes to 'offset' in JSON mode."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert response.model_dump(mode="json")["pagination_type"] == "offset"
+
+    def test_pagination_field_is_typed(self):
+        """pagination field is OffsetPagination, not the union."""
+        response = OffsetPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=OffsetPagination(
+                total_count=10, items_per_page=5, page=2, has_more=True
+            ),
+        )
+        assert isinstance(response.pagination, OffsetPagination)
+        assert response.pagination.total_count == 10
+        assert response.pagination.page == 2
+
+    def test_is_subclass_of_paginated_response(self):
+        """OffsetPaginatedResponse IS a PaginatedResponse."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert isinstance(response, PaginatedResponse)
+
+    def test_pagination_type_default_cannot_be_overridden_to_cursor(self):
+        """pagination_type rejects values other than OFFSET."""
+        with pytest.raises(ValidationError):
+            OffsetPaginatedResponse(
+                data=[],
+                pagination=OffsetPagination(
+                    total_count=0, items_per_page=10, page=1, has_more=False
+                ),
+                pagination_type=PaginationType.CURSOR,  # type: ignore[arg-type]
+            )
+
+    def test_filter_attributes_defaults_to_none(self):
+        """filter_attributes defaults to None."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert response.filter_attributes is None
+
+    def test_full_serialization(self):
+        """Full JSON serialization includes all expected fields."""
+        response = OffsetPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=OffsetPagination(
+                total_count=1, items_per_page=10, page=1, has_more=False
+            ),
+            filter_attributes={"status": ["active"]},
+        )
+        data = response.model_dump(mode="json")
+
+        assert data["pagination_type"] == "offset"
+        assert data["status"] == "SUCCESS"
+        assert data["data"] == [{"id": 1}]
+        assert data["pagination"]["total_count"] == 1
+        assert data["filter_attributes"] == {"status": ["active"]}
+
+
+class TestCursorPaginatedResponse:
+    """Tests for CursorPaginatedResponse schema."""
+
+    def test_pagination_type_is_cursor(self):
+        """pagination_type is always PaginationType.CURSOR."""
+        response = CursorPaginatedResponse(
+            data=[],
+            pagination=CursorPagination(
+                next_cursor=None, items_per_page=10, has_more=False
+            ),
+        )
+        assert response.pagination_type is PaginationType.CURSOR
+
+    def test_pagination_type_serializes_to_string(self):
+        """pagination_type serializes to 'cursor' in JSON mode."""
+        response = CursorPaginatedResponse(
+            data=[],
+            pagination=CursorPagination(
+                next_cursor=None, items_per_page=10, has_more=False
+            ),
+        )
+        assert response.model_dump(mode="json")["pagination_type"] == "cursor"
+
+    def test_pagination_field_is_typed(self):
+        """pagination field is CursorPagination, not the union."""
+        response = CursorPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=CursorPagination(
+                next_cursor="abc123",
+                prev_cursor=None,
+                items_per_page=20,
+                has_more=True,
+            ),
+        )
+        assert isinstance(response.pagination, CursorPagination)
+        assert response.pagination.next_cursor == "abc123"
+        assert response.pagination.has_more is True
+
+    def test_is_subclass_of_paginated_response(self):
+        """CursorPaginatedResponse IS a PaginatedResponse."""
+        response = CursorPaginatedResponse(
+            data=[],
+            pagination=CursorPagination(
+                next_cursor=None, items_per_page=10, has_more=False
+            ),
+        )
+        assert isinstance(response, PaginatedResponse)
+
+    def test_pagination_type_default_cannot_be_overridden_to_offset(self):
+        """pagination_type rejects values other than CURSOR."""
+        with pytest.raises(ValidationError):
+            CursorPaginatedResponse(
+                data=[],
+                pagination=CursorPagination(
+                    next_cursor=None, items_per_page=10, has_more=False
+                ),
+                pagination_type=PaginationType.OFFSET,  # type: ignore[arg-type]
+            )
+
+    def test_full_serialization(self):
+        """Full JSON serialization includes all expected fields."""
+        response = CursorPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=CursorPagination(
+                next_cursor="tok_next",
+                prev_cursor="tok_prev",
+                items_per_page=10,
+                has_more=True,
+            ),
+        )
+        data = response.model_dump(mode="json")
+
+        assert data["pagination_type"] == "cursor"
+        assert data["status"] == "SUCCESS"
+        assert data["pagination"]["next_cursor"] == "tok_next"
+        assert data["pagination"]["prev_cursor"] == "tok_prev"
+
+
 class TestFromAttributes:
    """Tests for from_attributes config (ORM mode)."""