fix: prev_cursor does not navigate backward (#136)

src/fastapi_toolsets/crud/factory.py

@@ -42,14 +42,17 @@ from .search import (
 )
 
 
-def _encode_cursor(value: Any) -> str:
+def _encode_cursor(value: Any, *, direction: str = "next") -> str:
-    """Encode cursor column value as an base64 string."""
+    """Encode a cursor column value and navigation direction as a base64 string."""
-    return base64.b64encode(json.dumps(str(value)).encode()).decode()
+    return base64.b64encode(
+        json.dumps({"val": str(value), "dir": direction}).encode()
+    ).decode()
 
 
-def _decode_cursor(cursor: str) -> str:
+def _decode_cursor(cursor: str) -> tuple[str, str]:
-    """Decode cursor base64 string."""
+    """Decode a cursor base64 string into ``(raw_value, direction)``."""
-    return json.loads(base64.b64decode(cursor.encode()).decode())
+    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,6 +1061,9 @@ class AsyncCrud(Generic[ModelType]):
                     "Supported types: Integer, BigInteger, SmallInteger, Uuid, "
                     "DateTime, Date, Float, Numeric."
                 )
+            if direction == "prev":
+                filters.append(cursor_column < cursor_val)
+            else:
                 filters.append(cursor_column > cursor_val)
 
         # Build search filters
@@ -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
+        # 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
+        # Restore ascending order when traversing backward
-        next_cursor: str | None = None
+        if direction == "prev":
-        if has_more and items_page:
+            items_page = list(reversed(items_page))
-            next_cursor = _encode_cursor(getattr(items_page[-1], cursor_col_name))
 
-        # 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:
+        if direction == "next" and cursor is not None and items_page:
-            prev_cursor = _encode_cursor(getattr(items_page[0], cursor_col_name))
+            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]
 

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):
+    async def test_prev_cursor_navigates_back(self, db_session: AsyncSession):
-        """prev_cursor encodes the value of the first item on the current page."""
+        """prev_cursor on page 2 navigates back to the same items as page 1."""
-        import base64
-        import json
-
         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
+        # Using prev_cursor should return the same items as page 1
-        decoded = json.loads(
+        back_to_page1 = await RoleCursorCrud.cursor_paginate(
-            base64.b64decode(page2.pagination.prev_cursor.encode()).decode()
+            db_session,
+            cursor=page2.pagination.prev_cursor,
+            items_per_page=5,
+            schema=RoleRead,
         )
-        first_item_id = str(page2.data[0].id)
+        assert isinstance(back_to_page1.pagination, CursorPagination)
-        assert decoded == first_item_id
+        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: