"""Pagination utilities and standardized query parameter handling. This module provides reusable utilities for implementing consistent pagination across all endpoints. All paginated endpoints should use these utilities to ensure a uniform API contract. Supported Pagination Modes: 1. **Offset-Based (Legacy)** — Uses page number + page_size. Query parameters: page (1-based), page_size (1-500) ⚠️ Performance degrades on large offsets (OFFSET requires scanning N rows). Use for: Small datasets, where performance is not critical. 2. **Cursor-Based (Recommended for large tables)** — Uses keyset pagination. Query parameters: cursor (opaque token for next/prev), page_size ✓ Constant-time performance regardless of dataset size. Use for: Large tables (>100K rows), paginated lists with sorting. Usage Examples: **Offset pagination (legacy):** ```python from app.utils.pagination import PAGINATION_DEFAULTS, create_pagination_metadata @router.get("/items") async def get_items( page: int = Query(default=PAGINATION_DEFAULTS["page"], ge=1), page_size: int = Query( default=PAGINATION_DEFAULTS["page_size"], ge=1, le=PAGINATION_DEFAULTS["max_page_size"], ), ): items = [...] total = 100 pagination = create_pagination_metadata(total, page, page_size) return MyListResponse(items=items, pagination=pagination) ``` **Cursor pagination (recommended):** ```python from app.utils.pagination import decode_cursor, encode_cursor, PAGINATION_DEFAULTS @router.get("/items") async def get_items( cursor: str | None = Query(None), page_size: int = Query( default=PAGINATION_DEFAULTS["page_size"], ge=1, le=PAGINATION_DEFAULTS["max_page_size"], ), ): # Decode cursor to get last_row_id last_row_id = decode_cursor(cursor) if cursor else None # Fetch items using keyset pagination (WHERE id > last_row_id) items, has_more = await repo.get_items_keyset(page_size, last_row_id) # Encode cursor for next page (last item's ID) next_cursor = encode_cursor(items[-1]["id"]) if items and has_more else None pagination = create_keyset_pagination_metadata(items, next_cursor, page_size) return MyListResponse(items=items, pagination=pagination) ``` """ import base64 import json from typing import TYPE_CHECKING, Final if TYPE_CHECKING: from app.models.response import PaginationMetadata __all__ = [ "PAGINATION_DEFAULTS", "get_offset", "compute_total_pages", "create_pagination_metadata", "encode_cursor", "decode_cursor", "create_keyset_pagination_metadata", ] # Standardized pagination defaults PAGINATION_DEFAULTS: Final[dict[str, int]] = { "page": 1, "page_size": 100, "max_page_size": 500, } def get_offset(page: int, page_size: int) -> int: """Calculate the database offset for a given page and page size. Args: page: 1-based page number. page_size: Items per page. Returns: 0-based database offset (number of items to skip). Raises: ValueError: If page or page_size is invalid (< 1). Example: ```python # Page 1, size 10 → offset 0 assert get_offset(1, 10) == 0 # Page 2, size 10 → offset 10 assert get_offset(2, 10) == 10 # Page 3, size 50 → offset 100 assert get_offset(3, 50) == 100 ``` """ if page < 1: raise ValueError(f"page must be >= 1, got {page}") if page_size < 1: raise ValueError(f"page_size must be >= 1, got {page_size}") return (page - 1) * page_size def compute_total_pages(total: int, page_size: int) -> int: """Calculate the total number of pages needed. Args: total: Total number of items across all pages. page_size: Items per page. Returns: The number of pages required to hold all items. Always at least 1, even when total is 0 (an empty page is still a page). Raises: ValueError: If page_size is invalid (< 1). Example: ```python assert compute_total_pages(0, 10) == 1 assert compute_total_pages(10, 10) == 1 assert compute_total_pages(11, 10) == 2 assert compute_total_pages(25, 5) == 5 ``` """ if page_size < 1: raise ValueError(f"page_size must be >= 1, got {page_size}") if total == 0: return 1 # Ceiling division: (total + page_size - 1) // page_size return (total + page_size - 1) // page_size def create_pagination_metadata(total: int, page: int, page_size: int) -> "PaginationMetadata": """Create pagination metadata with computed derived fields. This factory function computes all pagination-related information in a single place, ensuring consistency across all paginated endpoints. Args: total: Total number of items across all pages. page: Current page number (1-based). page_size: Items per page. Returns: :class:`~app.models.response.PaginationMetadata` with computed fields: - total_pages: Computed total number of pages needed - has_next_page: Whether there is a next page - has_prev_page: Whether there is a previous page Example: ```python metadata = create_pagination_metadata(total=150, page=2, page_size=50) assert metadata.total_pages == 3 assert metadata.has_next_page is True assert metadata.has_prev_page is True ``` """ from app.models.response import PaginationMetadata total_pages = compute_total_pages(total, page_size) has_next_page = page < total_pages has_prev_page = page > 1 return PaginationMetadata( page=page, page_size=page_size, total=total, total_pages=total_pages, has_next_page=has_next_page, has_prev_page=has_prev_page, ) # --------------------------------------------------------------------------- # Cursor-Based Pagination Functions # --------------------------------------------------------------------------- def encode_cursor(row_id: int) -> str: """Encode a row ID into an opaque cursor token. The cursor is a base64-encoded JSON object containing the row ID. This format is opaque to the client and must not be modified manually. Args: row_id: The database row ID to encode. Returns: Base64-encoded cursor string that can be passed to decode_cursor(). Raises: ValueError: If row_id is invalid (< 1). Example: ```python cursor = encode_cursor(42) assert isinstance(cursor, str) assert decode_cursor(cursor) == 42 ``` """ if row_id < 1: raise ValueError(f"row_id must be >= 1, got {row_id}") cursor_data = {"id": row_id} json_str = json.dumps(cursor_data, separators=(",", ":")) return base64.b64encode(json_str.encode()).decode("ascii") def decode_cursor(cursor: str) -> int: """Decode an opaque cursor token to retrieve the row ID. Decodes a base64-encoded JSON object containing the row ID. This is the inverse of encode_cursor(). Args: cursor: Cursor string produced by encode_cursor(). Returns: The row ID stored in the cursor. Raises: ValueError: If cursor is invalid (not base64-decodable or missing 'id' field). Example: ```python cursor = encode_cursor(42) assert decode_cursor(cursor) == 42 ``` """ try: json_str = base64.b64decode(cursor.encode("ascii")).decode("utf-8") cursor_data = json.loads(json_str) row_id = cursor_data.get("id") if not isinstance(row_id, int) or row_id < 1: raise ValueError(f"Invalid cursor: 'id' field must be an integer >= 1, got {row_id}") return row_id except (ValueError, TypeError, json.JSONDecodeError) as e: raise ValueError(f"Invalid cursor format: {e}") from e def create_keyset_pagination_metadata( items: list[dict[str, object]] | list[object], next_cursor: str | None, page_size: int, ) -> "PaginationMetadata": """Create pagination metadata for keyset (cursor-based) pagination. This function creates metadata for cursor-based pagination without the need to query the total row count. Frontend can determine if there are more pages by checking if the returned items count equals page_size. Args: items: The items returned from the keyset query (fetched count + 1). next_cursor: Cursor for fetching the next page, or None if no more pages. page_size: The requested page size. Returns: :class:`~app.models.response.PaginationMetadata` adapted for cursor pagination. Note: total and total_pages are set to -1 (unknown), has_prev_page is always False. Example: ```python items = await repo.get_items_keyset(page_size=10, last_row_id=None) metadata = create_keyset_pagination_metadata(items, next_cursor, page_size=10) assert metadata.total == -1 # Unknown in cursor pagination assert metadata.has_next_page == (next_cursor is not None) ``` """ from app.models.response import PaginationMetadata has_next_page = next_cursor is not None return PaginationMetadata( page=1, page_size=page_size, total=-1, total_pages=-1, has_next_page=has_next_page, has_prev_page=False, cursor=next_cursor, )