Coverage for mcpgateway / utils / pagination.py: 100%
176 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 03:05 +0000
1# -*- coding: utf-8 -*-
2"""Location: ./mcpgateway/utils/pagination.py
3Copyright 2025
4SPDX-License-Identifier: Apache-2.0
5Authors: Mihai Criveti
7Pagination Utilities for ContextForge.
9This module provides utilities for implementing efficient pagination
10across all ContextForge endpoints, supporting both offset-based and
11cursor-based pagination strategies.
13Features:
14- Offset-based pagination for simple use cases (<10K records)
15- Cursor-based pagination for large datasets (>10K records)
16- Automatic strategy selection based on result set size
17- Navigation link generation
18- Query parameter parsing and validation
20Examples:
21 Basic usage with pagination query::
23 from mcpgateway.utils.pagination import paginate_query
24 from sqlalchemy import select
25 from mcpgateway.common.models import Tool
27 async def list_tools(db: Session):
28 query = select(Tool).where(Tool.enabled == True)
29 result = await paginate_query(
30 db=db,
31 query=query,
32 page=1,
33 per_page=50,
34 base_url="/admin/tools"
35 )
36 return result
37"""
39# Standard
40import base64
41import logging
42import math
43from typing import Any, Dict, List, Optional, Tuple, Union
44from urllib.parse import urlencode
46# Third-Party
47from fastapi import Request
48import orjson
49from sqlalchemy import and_, func, or_, select
50from sqlalchemy.orm import Session
51from sqlalchemy.sql import Select
53# First-Party
54from mcpgateway.config import settings
55from mcpgateway.schemas import PaginationLinks, PaginationMeta
57logger = logging.getLogger(__name__)
60def encode_cursor(data: Dict[str, Any]) -> str:
61 """Encode pagination cursor data to base64.
63 Args:
64 data: Dictionary containing cursor data (id, created_at, etc.)
66 Returns:
67 Base64-encoded cursor string
69 Examples:
70 >>> cursor_data = {"id": "tool-123", "created_at": "2025-01-15T10:30:00Z"}
71 >>> cursor = encode_cursor(cursor_data)
72 >>> isinstance(cursor, str)
73 True
74 >>> len(cursor) > 0
75 True
77 >>> # Test with simple ID-only cursor
78 >>> simple_cursor = encode_cursor({"id": 42})
79 >>> isinstance(simple_cursor, str)
80 True
81 >>> len(simple_cursor) > 0
82 True
84 >>> # Test empty dict
85 >>> empty_cursor = encode_cursor({})
86 >>> isinstance(empty_cursor, str)
87 True
89 >>> # Test with numeric values
90 >>> numeric_cursor = encode_cursor({"id": 12345, "offset": 100})
91 >>> len(numeric_cursor) > 0
92 True
93 """
94 json_bytes = orjson.dumps(data, default=str, option=orjson.OPT_SORT_KEYS)
95 return base64.urlsafe_b64encode(json_bytes).decode()
98def decode_cursor(cursor: str) -> Dict[str, Any]:
99 """Decode pagination cursor from base64.
101 Args:
102 cursor: Base64-encoded cursor string
104 Returns:
105 Decoded cursor data dictionary
107 Raises:
108 ValueError: If cursor is invalid
110 Examples:
111 >>> cursor_data = {"id": "tool-123", "created_at": "2025-01-15T10:30:00Z"}
112 >>> cursor = encode_cursor(cursor_data)
113 >>> decoded = decode_cursor(cursor)
114 >>> decoded["id"]
115 'tool-123'
117 >>> # Test round-trip with numeric ID
118 >>> data = {"id": 42}
119 >>> encoded = encode_cursor(data)
120 >>> decoded = decode_cursor(encoded)
121 >>> decoded["id"]
122 42
124 >>> # Test with complex data
125 >>> complex_data = {"id": "abc-123", "page": 5, "filter": "active"}
126 >>> encoded_complex = encode_cursor(complex_data)
127 >>> decoded_complex = decode_cursor(encoded_complex)
128 >>> decoded_complex["id"]
129 'abc-123'
130 >>> decoded_complex["page"]
131 5
133 >>> # Test invalid cursor raises ValueError
134 >>> try:
135 ... decode_cursor("invalid-not-base64")
136 ... except ValueError as e:
137 ... "Invalid cursor" in str(e)
138 True
140 >>> # Test empty string raises ValueError
141 >>> try:
142 ... decode_cursor("")
143 ... except ValueError as e:
144 ... "Invalid cursor" in str(e)
145 True
146 """
147 try:
148 json_bytes = base64.urlsafe_b64decode(cursor.encode())
149 return orjson.loads(json_bytes)
150 except (ValueError, orjson.JSONDecodeError) as e:
151 raise ValueError(f"Invalid cursor: {e}")
154def generate_pagination_links(
155 base_url: str,
156 page: int,
157 per_page: int,
158 total_pages: int,
159 query_params: Optional[Dict[str, Any]] = None,
160 cursor: Optional[str] = None,
161 next_cursor: Optional[str] = None,
162 prev_cursor: Optional[str] = None,
163) -> PaginationLinks:
164 """Generate pagination navigation links.
166 Args:
167 base_url: Base URL for the endpoint
168 page: Current page number
169 per_page: Items per page
170 total_pages: Total number of pages
171 query_params: Additional query parameters to include
172 cursor: Current cursor (for cursor-based pagination)
173 next_cursor: Next page cursor
174 prev_cursor: Previous page cursor
176 Returns:
177 PaginationLinks object with navigation URLs
179 Examples:
180 >>> links = generate_pagination_links(
181 ... base_url="/admin/tools",
182 ... page=2,
183 ... per_page=50,
184 ... total_pages=5
185 ... )
186 >>> "/admin/tools?page=2" in links.self
187 True
188 >>> "/admin/tools?page=3" in links.next
189 True
191 >>> # Test first page
192 >>> first_page = generate_pagination_links(
193 ... base_url="/api/resources",
194 ... page=1,
195 ... per_page=25,
196 ... total_pages=10
197 ... )
198 >>> first_page.prev is None
199 True
200 >>> "/api/resources?page=2" in first_page.next
201 True
203 >>> # Test last page
204 >>> last_page = generate_pagination_links(
205 ... base_url="/api/prompts",
206 ... page=5,
207 ... per_page=20,
208 ... total_pages=5
209 ... )
210 >>> last_page.next is None
211 True
212 >>> "/api/prompts?page=4" in last_page.prev
213 True
215 >>> # Test cursor-based pagination
216 >>> cursor_links = generate_pagination_links(
217 ... base_url="/api/tools",
218 ... page=1,
219 ... per_page=50,
220 ... total_pages=0,
221 ... next_cursor="eyJpZCI6MTIzfQ=="
222 ... )
223 >>> "cursor=" in cursor_links.next
224 True
225 >>> "/api/tools?" in cursor_links.next
226 True
228 >>> # Test with query parameters
229 >>> links_with_params = generate_pagination_links(
230 ... base_url="/api/tools",
231 ... page=3,
232 ... per_page=100,
233 ... total_pages=10,
234 ... query_params={"filter": "active", "sort": "name"}
235 ... )
236 >>> "filter=active" in links_with_params.self
237 True
238 >>> "sort=name" in links_with_params.self
239 True
240 """
241 query_params = query_params or {}
243 def build_url(page_num: Optional[int] = None, cursor_val: Optional[str] = None) -> str:
244 """Build URL with query parameters.
246 Args:
247 page_num: Page number for offset pagination
248 cursor_val: Cursor value for cursor-based pagination
250 Returns:
251 str: Complete URL with query parameters
252 """
253 params = query_params.copy()
254 if cursor_val:
255 params["cursor"] = cursor_val
256 params["per_page"] = per_page
257 elif page_num is not None:
258 params["page"] = page_num
259 params["per_page"] = per_page
261 if params:
262 return f"{base_url}?{urlencode(params)}"
263 return base_url
265 # For cursor-based pagination
266 if cursor or next_cursor or prev_cursor:
267 return PaginationLinks(
268 self=build_url(cursor_val=cursor) if cursor else build_url(page_num=page),
269 first=build_url(page_num=1),
270 last=base_url, # Last page not applicable for cursor-based
271 next=build_url(cursor_val=next_cursor) if next_cursor else None,
272 prev=build_url(cursor_val=prev_cursor) if prev_cursor else None,
273 )
275 # For offset-based pagination
276 return PaginationLinks(
277 self=build_url(page_num=page),
278 first=build_url(page_num=1),
279 last=build_url(page_num=total_pages) if total_pages > 0 else build_url(page_num=1),
280 next=build_url(page_num=page + 1) if page < total_pages else None,
281 prev=build_url(page_num=page - 1) if page > 1 else None,
282 )
285async def offset_paginate(
286 db: Session,
287 query: Select,
288 page: int,
289 per_page: int,
290 base_url: str,
291 query_params: Optional[Dict[str, Any]] = None,
292 include_links: bool = True,
293 total_count: Optional[int] = None,
294) -> Dict[str, Any]:
295 """Paginate query using offset-based pagination.
297 Best for result sets < 10,000 records.
299 Args:
300 db: Database session
301 query: SQLAlchemy select query
302 page: Page number (1-indexed)
303 per_page: Items per page
304 base_url: Base URL for link generation
305 query_params: Additional query parameters
306 include_links: Whether to include navigation links
307 total_count: Pre-computed total count (avoids duplicate count query)
309 Returns:
310 Dictionary with 'data', 'pagination', and 'links' keys
312 Examples:
313 Basic offset pagination usage::
315 from mcpgateway.utils.pagination import offset_paginate
316 from sqlalchemy import select
317 from mcpgateway.common.models import Tool
319 async def list_tools_offset(db: Session, page: int = 1):
320 query = select(Tool).where(Tool.enabled == True)
321 result = await offset_paginate(
322 db=db,
323 query=query,
324 page=page,
325 per_page=50,
326 base_url="/admin/tools"
327 )
328 return result
329 """
330 # Validate parameters
331 page = max(1, page)
332 per_page = max(settings.pagination_min_page_size, min(per_page, settings.pagination_max_page_size))
334 # Get total count (use pre-computed count if provided to avoid duplicate queries)
335 if total_count is not None:
336 total_items = total_count
337 else:
338 count_query = select(func.count()).select_from(query.alias())
339 total_items = db.execute(count_query).scalar() or 0
341 # Calculate pagination metadata
342 total_pages = math.ceil(total_items / per_page) if total_items > 0 else 0
343 if total_pages > 0:
344 page = min(page, total_pages)
345 offset = (page - 1) * per_page
347 # Validate offset
348 if offset > settings.pagination_max_offset:
349 logger.warning(f"Offset {offset} exceeds maximum {settings.pagination_max_offset}")
350 offset = settings.pagination_max_offset
352 # Execute paginated query
353 paginated_query = query.offset(offset).limit(per_page)
354 items = db.execute(paginated_query).scalars().all()
356 # Build pagination metadata
357 pagination = PaginationMeta(
358 page=page,
359 per_page=per_page,
360 total_items=total_items,
361 total_pages=total_pages,
362 has_next=page < total_pages,
363 has_prev=page > 1,
364 next_cursor=None,
365 prev_cursor=None,
366 )
368 # Build links if requested
369 links = None
370 if include_links and settings.pagination_include_links:
371 links = generate_pagination_links(
372 base_url=base_url,
373 page=page,
374 per_page=per_page,
375 total_pages=total_pages,
376 query_params=query_params,
377 )
379 return {
380 "data": items,
381 "pagination": pagination,
382 "links": links,
383 }
386async def cursor_paginate(
387 db: Session,
388 query: Select,
389 cursor: Optional[str],
390 per_page: int,
391 base_url: str,
392 cursor_field: str = "created_at",
393 cursor_id_field: str = "id",
394 query_params: Optional[Dict[str, Any]] = None,
395 include_links: bool = True,
396 total_count: Optional[int] = None,
397) -> Dict[str, Any]:
398 """Paginate query using cursor-based pagination.
400 Best for result sets > 10,000 records. Uses keyset pagination
401 for consistent performance regardless of offset.
403 Args:
404 db: Database session
405 query: SQLAlchemy select query
406 cursor: Current cursor (None for first page)
407 per_page: Items per page
408 base_url: Base URL for link generation
409 cursor_field: Field to use for cursor (default: created_at)
410 cursor_id_field: ID field for tie-breaking (default: id)
411 query_params: Additional query parameters
412 include_links: Whether to include navigation links
413 total_count: Pre-computed total count (avoids duplicate count query)
415 Returns:
416 Dictionary with 'data', 'pagination', and 'links' keys
418 Examples:
419 Basic cursor pagination usage::
421 from mcpgateway.utils.pagination import cursor_paginate
422 from sqlalchemy import select
423 from mcpgateway.common.models import Tool
425 async def list_tools_cursor(db: Session, cursor: Optional[str] = None):
426 query = select(Tool).order_by(Tool.created_at.desc())
427 result = await cursor_paginate(
428 db=db,
429 query=query,
430 cursor=cursor,
431 per_page=50,
432 base_url="/admin/tools"
433 )
434 return result
435 """
436 # Validate parameters
437 per_page = max(settings.pagination_min_page_size, min(per_page, settings.pagination_max_page_size))
439 # Store unfiltered query for total count (before cursor filter)
440 unfiltered_query = query
442 # Decode cursor if provided
443 cursor_data = None
444 if cursor:
445 try:
446 cursor_data = decode_cursor(cursor)
447 except ValueError as e:
448 logger.warning(f"Invalid cursor: {e}")
449 cursor_data = None
451 # Apply cursor filter if provided
452 if cursor_data:
453 cursor_value = cursor_data.get(cursor_field)
454 cursor_id_value = cursor_data.get(cursor_id_field)
456 if cursor_value and cursor_id_value:
457 # Parse datetime strings for reliable comparisons
458 # Standard
459 from datetime import datetime
461 if isinstance(cursor_value, str):
462 try:
463 cursor_value = datetime.fromisoformat(cursor_value)
464 except (ValueError, TypeError):
465 pass # Keep as string if parsing fails
467 # Extract model class from query to access columns
468 entities = query.column_descriptions
469 if entities:
470 model = entities[0]["entity"]
471 cursor_col = getattr(model, cursor_field)
472 id_col = getattr(model, cursor_id_field)
474 # Keyset pagination for descending order (created_at DESC, id DESC)
475 # Filter: WHERE created_at < cursor_value OR (created_at = cursor_value AND id < cursor_id)
476 query = query.where(or_(cursor_col < cursor_value, and_(cursor_col == cursor_value, id_col < cursor_id_value)))
478 # Fetch one extra item to determine if there's a next page
479 paginated_query = query.limit(per_page + 1)
480 items = db.execute(paginated_query).scalars().all()
482 # Check if there are more items
483 has_next = len(items) > per_page
484 if has_next:
485 items = items[:per_page] # Remove the extra item
487 # Generate cursors
488 next_cursor = None
489 if has_next and items:
490 last_item = items[-1]
491 next_cursor = encode_cursor(
492 {
493 cursor_field: getattr(last_item, cursor_field, None),
494 cursor_id_field: getattr(last_item, cursor_id_field, None),
495 }
496 )
498 # Get total count (use pre-computed count if provided to avoid duplicate queries)
499 # Use unfiltered_query for count so total_items reflects the full dataset, not remaining items
500 if total_count is not None:
501 total_items = total_count
502 else:
503 count_query = select(func.count()).select_from(unfiltered_query.alias())
504 total_items = db.execute(count_query).scalar() or 0
506 # Build pagination metadata
507 pagination = PaginationMeta(
508 page=1, # Not applicable for cursor-based
509 per_page=per_page,
510 total_items=total_items,
511 total_pages=0, # Not applicable for cursor-based
512 has_next=has_next,
513 has_prev=cursor is not None,
514 next_cursor=next_cursor,
515 prev_cursor=None, # Implementing prev cursor requires bidirectional cursors
516 )
518 # Build links if requested
519 links = None
520 if include_links and settings.pagination_include_links:
521 links = generate_pagination_links(
522 base_url=base_url,
523 page=1,
524 per_page=per_page,
525 total_pages=0,
526 query_params=query_params,
527 cursor=cursor,
528 next_cursor=next_cursor,
529 prev_cursor=None,
530 )
532 return {
533 "data": items,
534 "pagination": pagination,
535 "links": links,
536 }
539async def paginate_query(
540 db: Session,
541 query: Select,
542 page: int = 1,
543 per_page: Optional[int] = None,
544 cursor: Optional[str] = None,
545 base_url: str = "",
546 query_params: Optional[Dict[str, Any]] = None,
547 use_cursor_threshold: bool = True,
548 total_count: Optional[int] = None,
549) -> Dict[str, Any]:
550 """Automatically paginate query using best strategy.
552 Selects between offset-based and cursor-based pagination
553 based on result set size and configuration.
555 Args:
556 db: Database session
557 query: SQLAlchemy select query
558 page: Page number (1-indexed)
559 per_page: Items per page (uses default if None)
560 cursor: Cursor for cursor-based pagination
561 base_url: Base URL for link generation
562 query_params: Additional query parameters
563 use_cursor_threshold: Whether to auto-switch to cursor-based
564 total_count: Pre-computed total count (avoids duplicate count query)
566 Returns:
567 Dictionary with 'data', 'pagination', and 'links' keys
569 Examples:
570 Automatic pagination with strategy selection::
572 from mcpgateway.utils.pagination import paginate_query
573 from sqlalchemy import select
574 from mcpgateway.common.models import Tool
576 async def list_tools_auto(db: Session, page: int = 1):
577 query = select(Tool)
578 # Automatically switches to cursor-based for large datasets
579 result = await paginate_query(
580 db=db,
581 query=query,
582 page=page,
583 base_url="/admin/tools"
584 )
585 # Result contains: data, pagination, links
586 return result
587 """
588 # Use default page size if not provided
589 if per_page is None:
590 per_page = settings.pagination_default_page_size
592 # If cursor is provided, use cursor-based pagination
593 if cursor and settings.pagination_cursor_enabled:
594 return await cursor_paginate(
595 db=db,
596 query=query,
597 cursor=cursor,
598 per_page=per_page,
599 base_url=base_url,
600 query_params=query_params,
601 total_count=total_count,
602 )
604 # Check if we should use cursor-based pagination based on total count
605 if use_cursor_threshold and settings.pagination_cursor_enabled:
606 # Use pre-computed count if provided, otherwise query for it
607 if total_count is not None:
608 total_items = total_count
609 else:
610 count_query = select(func.count()).select_from(query.alias())
611 total_items = db.execute(count_query).scalar() or 0
613 if total_items > settings.pagination_cursor_threshold:
614 logger.info(f"Switching to cursor-based pagination (total_items={total_items} > threshold={settings.pagination_cursor_threshold})")
615 # Pass pre-computed count to cursor_paginate to avoid duplicate query
616 return await cursor_paginate(
617 db=db,
618 query=query,
619 cursor=cursor,
620 per_page=per_page,
621 base_url=base_url,
622 query_params=query_params,
623 total_count=total_items,
624 )
626 # Pass pre-computed count to offset_paginate to avoid duplicate query
627 return await offset_paginate(
628 db=db,
629 query=query,
630 page=page,
631 per_page=per_page,
632 base_url=base_url,
633 query_params=query_params,
634 total_count=total_items,
635 )
637 # Use offset-based pagination (no threshold check was performed)
638 return await offset_paginate(
639 db=db,
640 query=query,
641 page=page,
642 per_page=per_page,
643 base_url=base_url,
644 query_params=query_params,
645 total_count=total_count,
646 )
649async def unified_paginate(
650 db: Session,
651 query: Select,
652 page: Optional[int] = None,
653 per_page: Optional[int] = None,
654 cursor: Optional[str] = None,
655 limit: Optional[int] = None,
656 base_url: str = "",
657 query_params: Optional[Dict[str, Any]] = None,
658) -> Union[Dict[str, Any], Tuple[List[Any], Optional[str]]]:
659 """Unified pagination helper that returns cursor or page format based on parameters.
661 This function eliminates duplication in service methods by handling both pagination
662 styles in one place. It automatically returns the appropriate format:
663 - Page-based format when `page` is provided: {"data": [...], "pagination": {...}, "links": {...}}
664 - Cursor-based format when `cursor` or neither is provided: (list, next_cursor)
666 Args:
667 db: Database session
668 query: SQLAlchemy select query (must include ORDER BY for cursor pagination)
669 page: Page number for page-based pagination (1-indexed)
670 per_page: Items per page for page-based pagination
671 cursor: Cursor for cursor-based pagination
672 limit: Maximum items for cursor-based pagination (overrides default page size)
673 base_url: Base URL for link generation in page-based mode
674 query_params: Additional query parameters for links
676 Returns:
677 Union[Dict[str, Any], Tuple[List[Any], Optional[str]]]:
678 If page is provided: Dict with {"data": [...], "pagination": {...}, "links": {...}}
679 Otherwise: tuple of (list, next_cursor) for backward compatibility
681 Examples:
682 >>> import asyncio
683 >>> from unittest.mock import MagicMock
684 >>>
685 >>> # Test cursor-based pagination returns tuple format (list, next_cursor)
686 >>> async def test_cursor_based():
687 ... mock_db = MagicMock()
688 ... mock_query = MagicMock()
689 ... mock_query.column_descriptions = []
690 ... mock_query.limit = MagicMock(return_value=mock_query)
691 ... mock_db.execute = MagicMock(return_value=MagicMock(
692 ... scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
693 ... ))
694 ... result = await unified_paginate(
695 ... db=mock_db,
696 ... query=mock_query,
697 ... cursor=None,
698 ... limit=50
699 ... )
700 ... return isinstance(result, tuple) and len(result) == 2
701 >>> asyncio.run(test_cursor_based())
702 True
703 >>>
704 >>> # Verify return type difference: cursor mode returns tuple, page mode returns dict
705 >>> # Note: Page-based mode testing requires complex SQLAlchemy mocking,
706 >>> # see unit tests in tests/ for comprehensive page-based pagination tests
707 """
709 # Determine page size
710 if per_page is None:
711 per_page = limit if limit and limit > 0 else settings.pagination_default_page_size
713 # PAGE-BASED PAGINATION
714 if page is not None:
715 # Use existing paginate_query for page-based
716 result = await paginate_query(
717 db=db,
718 query=query,
719 page=page,
720 per_page=per_page,
721 base_url=base_url,
722 query_params=query_params,
723 use_cursor_threshold=False, # Explicit page-based mode
724 )
726 return result
728 # CURSOR-BASED PAGINATION
729 # Determine page size from limit parameter
730 if limit is not None:
731 if limit == 0:
732 page_size = None # No limit, fetch all
733 else:
734 page_size = min(limit, settings.pagination_max_page_size)
735 else:
736 page_size = settings.pagination_default_page_size
738 # Decode cursor if provided
739 # Standard
740 from datetime import datetime
742 last_id = None
743 last_created = None
744 if cursor:
745 try:
746 cursor_data = decode_cursor(cursor)
747 last_id = cursor_data.get("id")
748 created_str = cursor_data.get("created_at")
749 if created_str:
750 last_created = datetime.fromisoformat(created_str)
751 except (ValueError, TypeError) as e:
752 logger.warning(f"Invalid cursor, ignoring: {e}")
754 # Apply cursor filter with keyset pagination (assumes query already has ORDER BY)
755 if last_id and last_created:
756 # Extract model from query to apply filters
757 entities = query.column_descriptions
758 if entities:
759 # Third-Party
760 from sqlalchemy import and_, or_
762 model = entities[0]["entity"]
763 # Assumes descending order: created_at DESC, id DESC
764 query = query.where(or_(model.created_at < last_created, and_(model.created_at == last_created, model.id < last_id)))
766 # Fetch page_size + 1 to determine if there are more results
767 if page_size is not None:
768 query = query.limit(page_size + 1)
769 items = db.execute(query).scalars().all()
771 # Check if there are more results
772 has_more = False
773 if page_size is not None:
774 has_more = len(items) > page_size
775 if has_more:
776 items = items[:page_size]
778 # Generate next_cursor if there are more results
779 next_cursor = None
780 if has_more and items:
781 last_item = items[-1]
782 item_created_at = getattr(last_item, "created_at", None)
783 item_id = getattr(last_item, "id", None)
785 # Warn if cursor fields are missing - keyset pagination won't work correctly
786 if item_id is None:
787 logger.warning(
788 f"Cursor pagination may not work correctly: model {type(last_item).__name__} has no 'id' field. "
789 "Subsequent pages may return duplicate results. Consider using page-based pagination instead."
790 )
792 cursor_data = {"created_at": item_created_at, "id": item_id}
793 # Handle datetime serialization
794 if cursor_data["created_at"]:
795 cursor_data["created_at"] = cursor_data["created_at"].isoformat()
796 next_cursor = encode_cursor(cursor_data)
798 return (items, next_cursor)
801def parse_pagination_params(request: Request) -> Dict[str, Any]:
802 """Parse pagination parameters from request.
804 Args:
805 request: FastAPI request object
807 Returns:
808 Dictionary with parsed pagination parameters
810 Examples:
811 >>> from fastapi import Request
812 >>> # Mock request with query params
813 >>> request = type('Request', (), {
814 ... 'query_params': {'page': '2', 'per_page': '100'}
815 ... })()
816 >>> params = parse_pagination_params(request)
817 >>> params['page']
818 2
819 >>> params['per_page']
820 100
822 >>> # Test with cursor
823 >>> request_with_cursor = type('Request', (), {
824 ... 'query_params': {'cursor': 'eyJpZCI6IDEyM30=', 'per_page': '25'}
825 ... })()
826 >>> params_cursor = parse_pagination_params(request_with_cursor)
827 >>> params_cursor['cursor']
828 'eyJpZCI6IDEyM30='
829 >>> params_cursor['per_page']
830 25
832 >>> # Test with sort parameters
833 >>> request_with_sort = type('Request', (), {
834 ... 'query_params': {'page': '1', 'sort_by': 'name', 'sort_order': 'asc'}
835 ... })()
836 >>> params_sort = parse_pagination_params(request_with_sort)
837 >>> params_sort['sort_by']
838 'name'
839 >>> params_sort['sort_order']
840 'asc'
842 >>> # Test with invalid page (negative) - should default to 1
843 >>> request_invalid = type('Request', (), {
844 ... 'query_params': {'page': '-5', 'per_page': '50'}
845 ... })()
846 >>> params_invalid = parse_pagination_params(request_invalid)
847 >>> params_invalid['page']
848 1
850 >>> # Test with no parameters - uses defaults
851 >>> request_empty = type('Request', (), {'query_params': {}})()
852 >>> params_empty = parse_pagination_params(request_empty)
853 >>> params_empty['page']
854 1
855 >>> 'cursor' in params_empty
856 True
857 >>> 'sort_by' in params_empty
858 True
859 """
860 page = int(request.query_params.get("page", 1))
861 per_page = int(request.query_params.get("per_page", settings.pagination_default_page_size))
862 cursor = request.query_params.get("cursor")
863 sort_by = request.query_params.get("sort_by", settings.pagination_default_sort_field)
864 sort_order = request.query_params.get("sort_order", settings.pagination_default_sort_order)
866 # Validate and constrain values
867 page = max(1, page)
868 per_page = max(settings.pagination_min_page_size, min(per_page, settings.pagination_max_page_size))
870 return {
871 "page": page,
872 "per_page": per_page,
873 "cursor": cursor,
874 "sort_by": sort_by,
875 "sort_order": sort_order,
876 }