From 0fb00d44daacc24c2b0aab32f288d23627e1f895 Mon Sep 17 00:00:00 2001 From: d3vyce <44915747+d3vyce@users.noreply.github.com> Date: Sat, 14 Mar 2026 17:18:53 +0100 Subject: [PATCH] fix: prev_cursor does not navigate backward (#136) --- src/fastapi_toolsets/crud/factory.py | 65 ++++++++++++++------ tests/test_crud.py | 88 ++++++++++++++++++++++++---- 2 files changed, 125 insertions(+), 28 deletions(-) diff --git a/src/fastapi_toolsets/crud/factory.py b/src/fastapi_toolsets/crud/factory.py index ce754ff..b91c88c 100644 --- a/src/fastapi_toolsets/crud/factory.py +++ b/src/fastapi_toolsets/crud/factory.py @@ -42,14 +42,17 @@ from .search import ( ) -def _encode_cursor(value: Any) -> str: - """Encode cursor column value as an base64 string.""" - return base64.b64encode(json.dumps(str(value)).encode()).decode() +def _encode_cursor(value: Any, *, direction: str = "next") -> str: + """Encode a cursor column value and navigation direction as a base64 string.""" + return base64.b64encode( + json.dumps({"val": str(value), "dir": direction}).encode() + ).decode() -def _decode_cursor(cursor: str) -> str: - """Decode cursor base64 string.""" - return json.loads(base64.b64decode(cursor.encode()).decode()) +def _decode_cursor(cursor: str) -> tuple[str, str]: + """Decode a cursor base64 string into ``(raw_value, direction)``.""" + payload = json.loads(base64.b64decode(cursor.encode()).decode()) + return payload["val"], payload["dir"] def _apply_joins(q: Any, joins: JoinType | None, outer_join: bool) -> Any: @@ -1038,8 +1041,9 @@ class AsyncCrud(Generic[ModelType]): cursor_column: Any = cls.cursor_column cursor_col_name: str = cursor_column.key + direction = "next" if cursor is not None: - raw_val = _decode_cursor(cursor) + raw_val, direction = _decode_cursor(cursor) col_type = cursor_column.property.columns[0].type if isinstance(col_type, Integer): cursor_val: Any = int(raw_val) @@ -1057,7 +1061,10 @@ class AsyncCrud(Generic[ModelType]): "Supported types: Integer, BigInteger, SmallInteger, Uuid, " "DateTime, Date, Float, Numeric." ) - filters.append(cursor_column > cursor_val) + if direction == "prev": + filters.append(cursor_column < cursor_val) + else: + filters.append(cursor_column > cursor_val) # Build search filters if search: @@ -1084,12 +1091,15 @@ class AsyncCrud(Generic[ModelType]): if resolved := cls._resolve_load_options(load_options): q = q.options(*resolved) - # Cursor column is always the primary sort - q = q.order_by(cursor_column) + # Cursor column is always the primary sort; reverse direction for prev traversal + if direction == "prev": + q = q.order_by(cursor_column.desc()) + else: + q = q.order_by(cursor_column) if order_by is not None: q = q.order_by(order_by) - # Fetch one extra to detect whether a next page exists + # Fetch one extra to detect whether another page exists in this direction q = q.limit(items_per_page + 1) result = await session.execute(q) raw_items = cast(list[ModelType], result.unique().scalars().all()) @@ -1097,15 +1107,34 @@ class AsyncCrud(Generic[ModelType]): has_more = len(raw_items) > items_per_page items_page = raw_items[:items_per_page] - # next_cursor points past the last item on this page - next_cursor: str | None = None - if has_more and items_page: - next_cursor = _encode_cursor(getattr(items_page[-1], cursor_col_name)) + # Restore ascending order when traversing backward + if direction == "prev": + items_page = list(reversed(items_page)) - # prev_cursor points to the first item on this page or None when on the first page + # next_cursor: points past the last item in ascending order + next_cursor: str | None = None + if direction == "next": + if has_more and items_page: + next_cursor = _encode_cursor( + getattr(items_page[-1], cursor_col_name), direction="next" + ) + else: + # Going backward: always provide a next_cursor to allow returning forward + if items_page: + next_cursor = _encode_cursor( + getattr(items_page[-1], cursor_col_name), direction="next" + ) + + # prev_cursor: points before the first item in ascending order prev_cursor: str | None = None - if cursor is not None and items_page: - prev_cursor = _encode_cursor(getattr(items_page[0], cursor_col_name)) + if direction == "next" and cursor is not None and items_page: + prev_cursor = _encode_cursor( + getattr(items_page[0], cursor_col_name), direction="prev" + ) + elif direction == "prev" and has_more and items_page: + prev_cursor = _encode_cursor( + getattr(items_page[0], cursor_col_name), direction="prev" + ) items: list[Any] = [schema.model_validate(item) for item in items_page] diff --git a/tests/test_crud.py b/tests/test_crud.py index 58e9a0f..076543f 100644 --- a/tests/test_crud.py +++ b/tests/test_crud.py @@ -1969,11 +1969,8 @@ class TestCursorPaginatePrevCursor: assert page2.pagination.prev_cursor is not None @pytest.mark.anyio - async def test_prev_cursor_points_to_first_item(self, db_session: AsyncSession): - """prev_cursor encodes the value of the first item on the current page.""" - import base64 - import json - + async def test_prev_cursor_navigates_back(self, db_session: AsyncSession): + """prev_cursor on page 2 navigates back to the same items as page 1.""" for i in range(10): await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}")) @@ -1992,12 +1989,83 @@ class TestCursorPaginatePrevCursor: assert isinstance(page2.pagination, CursorPagination) assert page2.pagination.prev_cursor is not None - # Decode prev_cursor and compare to first item's id - decoded = json.loads( - base64.b64decode(page2.pagination.prev_cursor.encode()).decode() + # Using prev_cursor should return the same items as page 1 + back_to_page1 = await RoleCursorCrud.cursor_paginate( + db_session, + cursor=page2.pagination.prev_cursor, + items_per_page=5, + schema=RoleRead, ) - first_item_id = str(page2.data[0].id) - assert decoded == first_item_id + assert isinstance(back_to_page1.pagination, CursorPagination) + assert [r.id for r in back_to_page1.data] == [r.id for r in page1.data] + assert back_to_page1.pagination.prev_cursor is None + + @pytest.mark.anyio + async def test_prev_cursor_empty_result_when_no_items_before( + self, db_session: AsyncSession + ): + """Going backward past the first item returns an empty page.""" + from fastapi_toolsets.crud.factory import _encode_cursor + from fastapi_toolsets.schemas import CursorPagination + + await IntRoleCursorCrud.create(db_session, IntRoleCreate(name="role00")) + + page1 = await IntRoleCursorCrud.cursor_paginate( + db_session, items_per_page=5, schema=IntRoleRead + ) + assert isinstance(page1.pagination, CursorPagination) + + # Manually craft a backward cursor before any existing id + before_all = _encode_cursor(0, direction="prev") + empty = await IntRoleCursorCrud.cursor_paginate( + db_session, cursor=before_all, items_per_page=5, schema=IntRoleRead + ) + + assert isinstance(empty.pagination, CursorPagination) + assert empty.data == [] + assert empty.pagination.next_cursor is None + assert empty.pagination.prev_cursor is None + + @pytest.mark.anyio + async def test_prev_cursor_set_when_more_pages_behind( + self, db_session: AsyncSession + ): + """Going backward on page 2 (of 3) still exposes a prev_cursor for page 1.""" + for i in range(9): + await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}")) + + from fastapi_toolsets.schemas import CursorPagination + + page1 = await RoleCursorCrud.cursor_paginate( + db_session, items_per_page=3, schema=RoleRead + ) + assert isinstance(page1.pagination, CursorPagination) + page2 = await RoleCursorCrud.cursor_paginate( + db_session, + cursor=page1.pagination.next_cursor, + items_per_page=3, + schema=RoleRead, + ) + assert isinstance(page2.pagination, CursorPagination) + page3 = await RoleCursorCrud.cursor_paginate( + db_session, + cursor=page2.pagination.next_cursor, + items_per_page=3, + schema=RoleRead, + ) + assert isinstance(page3.pagination, CursorPagination) + assert page3.pagination.prev_cursor is not None + + # Going back to page 2 should still have a prev_cursor pointing at page 1 + back_to_page2 = await RoleCursorCrud.cursor_paginate( + db_session, + cursor=page3.pagination.prev_cursor, + items_per_page=3, + schema=RoleRead, + ) + assert isinstance(back_to_page2.pagination, CursorPagination) + assert [r.id for r in back_to_page2.data] == [r.id for r in page2.data] + assert back_to_page2.pagination.prev_cursor is not None class TestCursorPaginateWithSearch: