feat: implement cursor pagination

fiqrikm18 · fiqrikm18 · commit 84fd245ee4e6 · 2026-06-19T18:20:52.000+07:00
diff --git a/src/core/schemas/response.py b/src/core/schemas/response.py
@@ -37,6 +37,30 @@ class PaginatedResponse(BaseModel, Generic[T]):
     data: List[T]
 
 
+# ==========================================
+# Cursor-Based Pagination
+# ==========================================
+class CursorMeta(BaseModel):
+    next_cursor: Optional[str] = Field(
+        default=None, description="Cursor for the next page. None if no more items."
+    )
+    prev_cursor: Optional[str] = Field(
+        default=None, description="Cursor for the previous page. None if at the start."
+    )
+    has_next: bool = Field(
+        ..., description="True if there are more items after this page"
+    )
+    has_prev: bool = Field(..., description="True if there are items before this page")
+    limit: int = Field(..., description="Number of items per page")
+
+
+class CursorPaginatedResponse(BaseModel, Generic[T]):
+    success: bool = Field(default=True)
+    message: str = Field(default="Cursor-paginated data retrieved successfully")
+    meta: CursorMeta
+    data: List[T]
+
+
 # ==========================================
 # 3. Standard Error Response
 # ==========================================
diff --git a/src/core/utils/cursor.py b/src/core/utils/cursor.py
@@ -0,0 +1,36 @@
+import base64
+import json
+from datetime import datetime
+from enum import Enum
+from uuid import UUID
+
+
+class CursorDirection(Enum):
+    DIRECTION_NEXT = "next"
+    DIRECTION_PREV = "prev"
+
+
+def encode_cursor(created_at: datetime, id: UUID, dir: CursorDirection) -> str:
+    """
+    Encode a cursor from timestamp and ID.
+    Format: base64(json({"t": "ISO_TIMESTAMP", "id": "UUID"}))
+    """
+    cursor_data = {"t": created_at.isoformat(), "id": str(id), "dir": dir.value}
+    json_str = json.dumps(cursor_data)
+    return base64.urlsafe_b64encode(json_str.encode()).decode()
+
+
+def decode_cursor(cursor: str) -> tuple[datetime, UUID, CursorDirection]:
+    """
+    Decode a cursor back to timestamp and ID.
+    Returns: (created_at, id)
+    """
+    try:
+        json_str = base64.urlsafe_b64decode(cursor.encode()).decode()
+        cursor_data = json.loads(json_str)
+        created_at = datetime.fromisoformat(cursor_data["t"])
+        dir = CursorDirection(cursor_data["dir"])
+        id = UUID(cursor_data["id"])
+        return created_at, id, dir
+    except Exception as e:
+        raise ValueError(f"Invalid cursor format: {e}")
diff --git a/src/modules/todo/application/list_todo/handler.py b/src/modules/todo/application/list_todo/handler.py
@@ -1,3 +1,6 @@
+from datetime import datetime
+from uuid import UUID
+
 from src.modules.todo.application.list_todo.query import GetTodosQuery
 from src.modules.todo.application.list_todo.validation import validate_get_todos_query
 from src.modules.todo.domain.entities.todo import Todo
@@ -12,3 +15,27 @@ async def execute(self, command: GetTodosQuery) -> list[Todo]:
         validate_get_todos_query(command)
 
         return await self.todo_repo.get_all_by_user(command.user_id)
+
+
+class GetTodosCursorQuery:
+    def __init__(self, todo_repo: TodoRepository):
+        self.todo_repo = todo_repo
+
+    async def execute(
+        self,
+        user_id: UUID,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: str = "next",
+    ) -> tuple[list[Todo], bool]:
+        """
+        Returns: (items, has_more)
+        """
+        return await self.todo_repo.get_by_user_cursor(
+            user_id=user_id,
+            cursor_created_at=cursor_created_at,
+            cursor_id=cursor_id,
+            limit=limit,
+            direction=direction,
+        )
diff --git a/src/modules/todo/domain/repositories/todo_repository.py b/src/modules/todo/domain/repositories/todo_repository.py
@@ -1,6 +1,8 @@
 from abc import ABC, abstractmethod
+from datetime import datetime
 from uuid import UUID
 
+from src.core.utils.cursor import CursorDirection
 from src.modules.todo.domain.entities.todo import Todo
 
 
@@ -13,6 +15,21 @@ async def get_by_id(self, todo_id: UUID) -> Todo | None:
     async def get_all_by_user(self, user_id: UUID) -> list[Todo]:
         pass
 
+    @abstractmethod
+    async def get_by_user_cursor(
+        self,
+        user_id: UUID,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Todo], bool]:
+        """
+        Get todos with cursor pagination.
+        Returns: (items, has_more)
+        """
+        pass
+
     @abstractmethod
     async def save(self, todo: Todo) -> Todo:
         pass
diff --git a/src/modules/todo/infrastructure/repositories/todo_repository.py b/src/modules/todo/infrastructure/repositories/todo_repository.py
@@ -1,8 +1,10 @@
+from datetime import datetime
 from uuid import UUID
 
-from sqlalchemy import delete, select
+from sqlalchemy import and_, delete, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.core.utils.cursor import CursorDirection
 from src.modules.todo.domain.entities.todo import Todo
 from src.modules.todo.domain.repositories.todo_repository import TodoRepository
 from src.modules.todo.infrastructure.models.todo_model import TodoModel
@@ -25,6 +27,70 @@ async def get_by_id(self, todo_id: UUID) -> Todo | None:
             user_id=model.user_id,
         )
 
+    async def get_by_user_cursor(
+        self,
+        user_id: UUID,
+        cursor_created_at: datetime | None = None,
+        cursor_id: UUID | None = None,
+        limit: int = 10,
+        direction: CursorDirection = CursorDirection.DIRECTION_NEXT,
+    ) -> tuple[list[Todo], bool]:
+        """
+        Cursor pagination logic:
+        - If direction="next": Get items AFTER the cursor (older items)
+        - If direction="prev": Get items BEFORE the cursor (newer items)
+        """
+        query = select(TodoModel).where(
+            TodoModel.user_id == user_id,
+            TodoModel.deleted_at.is_(None),
+        )
+
+        # Apply cursor filter if provided
+        if cursor_created_at and cursor_id:
+            if direction == CursorDirection.DIRECTION_NEXT:
+                # Get items older than cursor (created_at < cursor OR (created_at == cursor AND id < cursor_id))
+                query = query.where(
+                    or_(
+                        TodoModel.created_at < cursor_created_at,
+                        and_(
+                            TodoModel.created_at == cursor_created_at,
+                            TodoModel.id < cursor_id,
+                        ),
+                    )
+                )
+                query = query.order_by(TodoModel.created_at.desc(), TodoModel.id.desc())
+            else:
+                # Get items newer than cursor (created_at > cursor OR (created_at == cursor AND id > cursor_id))
+                query = query.where(
+                    or_(
+                        TodoModel.created_at > cursor_created_at,
+                        and_(
+                            TodoModel.created_at == cursor_created_at,
+                            TodoModel.id > cursor_id,
+                        ),
+                    )
+                )
+                query = query.order_by(TodoModel.created_at.asc(), TodoModel.id.asc())
+        else:
+            # No cursor, just get the first page
+            query = query.order_by(TodoModel.created_at.desc(), TodoModel.id.desc())
+
+        # Fetch limit + 1 to check if there are more items
+        query = query.limit(limit + 1)
+
+        result = await self.db.execute(query)
+        models = result.scalars().all()
+
+        # Check if there are more items
+        has_more = len(models) > limit
+        models = models[:limit]  # Trim to actual limit
+
+        # If we fetched "prev", reverse to maintain consistent order (newest first)
+        if direction == CursorDirection.DIRECTION_PREV:
+            models = list(reversed(models))
+
+        return [self._to_entity(m) for m in models], has_more
+
     async def get_all_by_user(self, user_id: UUID) -> list[Todo]:
         result = await self.db.execute(
             select(TodoModel).where(TodoModel.user_id == user_id)
diff --git a/src/modules/todo/presentation/dependency.py b/src/modules/todo/presentation/dependency.py
@@ -4,7 +4,9 @@
 from src.core.database.postgres.session import get_db, get_unit_of_work
 from src.modules.todo.application.create_todo.handler import CreateTodoHandler
 from src.modules.todo.application.delete_todo.handler import DeleteTodoHandler
-from src.modules.todo.application.list_todo.handler import GetTodosQueryHandler
+from src.modules.todo.application.list_todo.handler import (
+    GetTodosCursorQuery,
+)
 from src.modules.todo.application.update_todo.handler import UpdateTodoHandler
 from src.modules.todo.domain.repositories.todo_repository import TodoRepository
 from src.modules.todo.infrastructure.repositories.todo_repository import (
@@ -38,7 +40,7 @@ def get_delete_todo_handler(
     return DeleteTodoHandler(repo, unit_of_work)
 
 
-def get_get_todos_query_handler(
+def get_todos_query_handler(
     repo: TodoRepository = Depends(get_todo_repository),
-) -> GetTodosQueryHandler:
-    return GetTodosQueryHandler(repo)
+) -> GetTodosCursorQuery:
+    return GetTodosCursorQuery(repo)
diff --git a/src/modules/todo/presentation/routers/todo_router.py b/src/modules/todo/presentation/routers/todo_router.py
@@ -1,9 +1,8 @@
+from typing import Optional
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 
-from src.core.schemas.response import PaginatedResponse, SuccessResponse
-from src.modules.todo.presentation.schemas.response import TodoResponse
 from src.core.authorization.dependencies import require_permission
 from src.core.authorization.permissions import (
     CREATE_ACTION,
@@ -12,10 +11,18 @@
     TODO_RESOURCE,
     UPDATE_ACTION,
 )
+from src.core.schemas.response import (
+    CursorMeta,
+    CursorPaginatedResponse,
+    SuccessResponse,
+)
+from src.core.utils.cursor import CursorDirection, decode_cursor, encode_cursor
 from src.modules.todo.application.create_todo.command import CreateTodoCommand
 from src.modules.todo.application.create_todo.handler import CreateTodoHandler
 from src.modules.todo.application.delete_todo.handler import DeleteTodoHandler
-from src.modules.todo.application.list_todo.handler import GetTodosQueryHandler
+from src.modules.todo.application.list_todo.handler import (
+    GetTodosCursorQuery,
+)
 from src.modules.todo.application.list_todo.query import GetTodosQuery
 from src.modules.todo.application.update_todo.command import UpdateTodoCommand
 from src.modules.todo.application.update_todo.handler import UpdateTodoHandler
@@ -26,9 +33,10 @@
 from src.modules.todo.presentation.dependency import (
     get_create_todo_handler,
     get_delete_todo_handler,
-    get_get_todos_query_handler,
+    get_todos_query_handler,
     get_update_todo_handler,
 )
+from src.modules.todo.presentation.schemas.response import TodoResponse
 
 router = APIRouter(prefix="/todos", tags=["Todos"])
 
@@ -55,24 +63,68 @@ async def create_todo(
     )
 
 
-@router.get("/", response_model=PaginatedResponse[TodoResponse])
+@router.get("/", response_model=CursorPaginatedResponse[TodoResponse])
 async def get_todos(
+    cursor: Optional[str] = Query(
+        None, description="Cursor for pagination (from previous response)"
+    ),
+    limit: int = Query(10, ge=1, le=100, description="Number of items per page"),
     current_user: dict = Depends(require_permission(TODO_RESOURCE, READ_ACTION)),
-    query: GetTodosQueryHandler = Depends(get_get_todos_query_handler),
+    query: GetTodosCursorQuery = Depends(get_todos_query_handler),
 ):
+    cursor_created_at = None
+    cursor_id = None
+    if cursor:
+        cursor_created_at, cursor_id, direction = decode_cursor(cursor)
+
     command = GetTodosQuery(user_id=current_user.get("id"))
-    todos = await query.execute(command=command)
-    return PaginatedResponse(
-        message="fetch todo success",
-        success=True,
-        data=[
-            TodoResponse(
-                id=str(todo.id),
-                title=todo.title,
-                is_completed=todo.is_completed,
-            )
-            for todo in todos
-        ],
+    todos, has_more = await query.execute(
+        user_id=command.user_id,
+        cursor_created_at=cursor_created_at,
+        cursor_id=cursor_id,
+        limit=limit,
+        direction=direction,
+    )
+    response_todos = [
+        TodoResponse(
+            id=str(t.id),
+            title=t.title,
+            description=t.description,
+            is_completed=t.is_completed,
+            created_at=t.created_at.isoformat(),
+        )
+        for t in todos
+    ]
+
+    next_cursor = None
+    prev_cursor = None
+
+    if has_more and len(todos) > 0:
+        last_item = todos[-1]
+        next_cursor = encode_cursor(
+            last_item.created_at,
+            last_item.id,
+            CursorDirection.DIRECTION_NEXT,
+        )
+
+    if cursor and len(todos) > 0:
+        first_item = todos[0]
+        prev_cursor = encode_cursor(
+            first_item.created_at,
+            first_item.id,
+            CursorDirection.DIRECTION_PREV,
+        )
+
+    return CursorPaginatedResponse(
+        message="Todos retrieved successfully",
+        meta=CursorMeta(
+            next_cursor=next_cursor,
+            prev_cursor=prev_cursor,
+            has_next=has_more,
+            has_prev=cursor is not None,
+            limit=limit,
+        ),
+        data=response_todos,
     )
 
 
