Coverage for mcpgateway / utils / pagination.py: 100%
174 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-02-11 07:10 +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 MCP Gateway.
9This module provides utilities for implementing efficient pagination
10across all MCP Gateway 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 offset = (page - 1) * per_page
345 # Validate offset
346 if offset > settings.pagination_max_offset:
347 logger.warning(f"Offset {offset} exceeds maximum {settings.pagination_max_offset}")
348 offset = settings.pagination_max_offset
350 # Execute paginated query
351 paginated_query = query.offset(offset).limit(per_page)
352 items = db.execute(paginated_query).scalars().all()
354 # Build pagination metadata
355 pagination = PaginationMeta(
356 page=page,
357 per_page=per_page,
358 total_items=total_items,
359 total_pages=total_pages,
360 has_next=page < total_pages,
361 has_prev=page > 1,
362 next_cursor=None,
363 prev_cursor=None,
364 )
366 # Build links if requested
367 links = None
368 if include_links and settings.pagination_include_links:
369 links = generate_pagination_links(
370 base_url=base_url,
371 page=page,
372 per_page=per_page,
373 total_pages=total_pages,
374 query_params=query_params,
375 )
377 return {
378 "data": items,
379 "pagination": pagination,
380 "links": links,
381 }
384async def cursor_paginate(
385 db: Session,
386 query: Select,
387 cursor: Optional[str],
388 per_page: int,
389 base_url: str,
390 cursor_field: str = "created_at",
391 cursor_id_field: str = "id",
392 query_params: Optional[Dict[str, Any]] = None,
393 include_links: bool = True,
394 total_count: Optional[int] = None,
395) -> Dict[str, Any]:
396 """Paginate query using cursor-based pagination.
398 Best for result sets > 10,000 records. Uses keyset pagination
399 for consistent performance regardless of offset.
401 Args:
402 db: Database session
403 query: SQLAlchemy select query
404 cursor: Current cursor (None for first page)
405 per_page: Items per page
406 base_url: Base URL for link generation
407 cursor_field: Field to use for cursor (default: created_at)
408 cursor_id_field: ID field for tie-breaking (default: id)
409 query_params: Additional query parameters
410 include_links: Whether to include navigation links
411 total_count: Pre-computed total count (avoids duplicate count query)
413 Returns:
414 Dictionary with 'data', 'pagination', and 'links' keys
416 Examples:
417 Basic cursor pagination usage::
419 from mcpgateway.utils.pagination import cursor_paginate
420 from sqlalchemy import select
421 from mcpgateway.common.models import Tool
423 async def list_tools_cursor(db: Session, cursor: Optional[str] = None):
424 query = select(Tool).order_by(Tool.created_at.desc())
425 result = await cursor_paginate(
426 db=db,
427 query=query,
428 cursor=cursor,
429 per_page=50,
430 base_url="/admin/tools"
431 )
432 return result
433 """
434 # Validate parameters
435 per_page = max(settings.pagination_min_page_size, min(per_page, settings.pagination_max_page_size))
437 # Store unfiltered query for total count (before cursor filter)
438 unfiltered_query = query
440 # Decode cursor if provided
441 cursor_data = None
442 if cursor:
443 try:
444 cursor_data = decode_cursor(cursor)
445 except ValueError as e:
446 logger.warning(f"Invalid cursor: {e}")
447 cursor_data = None
449 # Apply cursor filter if provided
450 if cursor_data:
451 cursor_value = cursor_data.get(cursor_field)
452 cursor_id_value = cursor_data.get(cursor_id_field)
454 if cursor_value and cursor_id_value:
455 # Parse datetime strings for reliable comparisons
456 # Standard
457 from datetime import datetime
459 if isinstance(cursor_value, str):
460 try:
461 cursor_value = datetime.fromisoformat(cursor_value)
462 except (ValueError, TypeError):
463 pass # Keep as string if parsing fails
465 # Extract model class from query to access columns
466 entities = query.column_descriptions
467 if entities:
468 model = entities[0]["entity"]
469 cursor_col = getattr(model, cursor_field)
470 id_col = getattr(model, cursor_id_field)
472 # Keyset pagination for descending order (created_at DESC, id DESC)
473 # Filter: WHERE created_at < cursor_value OR (created_at = cursor_value AND id < cursor_id)
474 query = query.where(or_(cursor_col < cursor_value, and_(cursor_col == cursor_value, id_col < cursor_id_value)))
476 # Fetch one extra item to determine if there's a next page
477 paginated_query = query.limit(per_page + 1)
478 items = db.execute(paginated_query).scalars().all()
480 # Check if there are more items
481 has_next = len(items) > per_page
482 if has_next:
483 items = items[:per_page] # Remove the extra item
485 # Generate cursors
486 next_cursor = None
487 if has_next and items:
488 last_item = items[-1]
489 next_cursor = encode_cursor(
490 {
491 cursor_field: getattr(last_item, cursor_field, None),
492 cursor_id_field: getattr(last_item, cursor_id_field, None),
493 }
494 )
496 # Get total count (use pre-computed count if provided to avoid duplicate queries)
497 # Use unfiltered_query for count so total_items reflects the full dataset, not remaining items
498 if total_count is not None:
499 total_items = total_count
500 else:
501 count_query = select(func.count()).select_from(unfiltered_query.alias())
502 total_items = db.execute(count_query).scalar() or 0
504 # Build pagination metadata
505 pagination = PaginationMeta(
506 page=1, # Not applicable for cursor-based
507 per_page=per_page,
508 total_items=total_items,
509 total_pages=0, # Not applicable for cursor-based
510 has_next=has_next,
511 has_prev=cursor is not None,
512 next_cursor=next_cursor,
513 prev_cursor=None, # Implementing prev cursor requires bidirectional cursors
514 )
516 # Build links if requested
517 links = None
518 if include_links and settings.pagination_include_links:
519 links = generate_pagination_links(
520 base_url=base_url,
521 page=1,
522 per_page=per_page,
523 total_pages=0,
524 query_params=query_params,
525 cursor=cursor,
526 next_cursor=next_cursor,
527 prev_cursor=None,
528 )
530 return {
531 "data": items,
532 "pagination": pagination,
533 "links": links,
534 }
537async def paginate_query(
538 db: Session,
539 query: Select,
540 page: int = 1,
541 per_page: Optional[int] = None,
542 cursor: Optional[str] = None,
543 base_url: str = "",
544 query_params: Optional[Dict[str, Any]] = None,
545 use_cursor_threshold: bool = True,
546 total_count: Optional[int] = None,
547) -> Dict[str, Any]:
548 """Automatically paginate query using best strategy.
550 Selects between offset-based and cursor-based pagination
551 based on result set size and configuration.
553 Args:
554 db: Database session
555 query: SQLAlchemy select query
556 page: Page number (1-indexed)
557 per_page: Items per page (uses default if None)
558 cursor: Cursor for cursor-based pagination
559 base_url: Base URL for link generation
560 query_params: Additional query parameters
561 use_cursor_threshold: Whether to auto-switch to cursor-based
562 total_count: Pre-computed total count (avoids duplicate count query)
564 Returns:
565 Dictionary with 'data', 'pagination', and 'links' keys
567 Examples:
568 Automatic pagination with strategy selection::
570 from mcpgateway.utils.pagination import paginate_query
571 from sqlalchemy import select
572 from mcpgateway.common.models import Tool
574 async def list_tools_auto(db: Session, page: int = 1):
575 query = select(Tool)
576 # Automatically switches to cursor-based for large datasets
577 result = await paginate_query(
578 db=db,
579 query=query,
580 page=page,
581 base_url="/admin/tools"
582 )
583 # Result contains: data, pagination, links
584 return result
585 """
586 # Use default page size if not provided
587 if per_page is None:
588 per_page = settings.pagination_default_page_size
590 # If cursor is provided, use cursor-based pagination
591 if cursor and settings.pagination_cursor_enabled:
592 return await cursor_paginate(
593 db=db,
594 query=query,
595 cursor=cursor,
596 per_page=per_page,
597 base_url=base_url,
598 query_params=query_params,
599 total_count=total_count,
600 )
602 # Check if we should use cursor-based pagination based on total count
603 if use_cursor_threshold and settings.pagination_cursor_enabled:
604 # Use pre-computed count if provided, otherwise query for it
605 if total_count is not None:
606 total_items = total_count
607 else:
608 count_query = select(func.count()).select_from(query.alias())
609 total_items = db.execute(count_query).scalar() or 0
611 if total_items > settings.pagination_cursor_threshold:
612 logger.info(f"Switching to cursor-based pagination (total_items={total_items} > threshold={settings.pagination_cursor_threshold})")
613 # Pass pre-computed count to cursor_paginate to avoid duplicate query
614 return await cursor_paginate(
615 db=db,
616 query=query,
617 cursor=cursor,
618 per_page=per_page,
619 base_url=base_url,
620 query_params=query_params,
621 total_count=total_items,
622 )
624 # Pass pre-computed count to offset_paginate to avoid duplicate query
625 return await offset_paginate(
626 db=db,
627 query=query,
628 page=page,
629 per_page=per_page,
630 base_url=base_url,
631 query_params=query_params,
632 total_count=total_items,
633 )
635 # Use offset-based pagination (no threshold check was performed)
636 return await offset_paginate(
637 db=db,
638 query=query,
639 page=page,
640 per_page=per_page,
641 base_url=base_url,
642 query_params=query_params,
643 total_count=total_count,
644 )
647async def unified_paginate(
648 db: Session,
649 query: Select,
650 page: Optional[int] = None,
651 per_page: Optional[int] = None,
652 cursor: Optional[str] = None,
653 limit: Optional[int] = None,
654 base_url: str = "",
655 query_params: Optional[Dict[str, Any]] = None,
656) -> Union[Dict[str, Any], Tuple[List[Any], Optional[str]]]:
657 """Unified pagination helper that returns cursor or page format based on parameters.
659 This function eliminates duplication in service methods by handling both pagination
660 styles in one place. It automatically returns the appropriate format:
661 - Page-based format when `page` is provided: {"data": [...], "pagination": {...}, "links": {...}}
662 - Cursor-based format when `cursor` or neither is provided: (list, next_cursor)
664 Args:
665 db: Database session
666 query: SQLAlchemy select query (must include ORDER BY for cursor pagination)
667 page: Page number for page-based pagination (1-indexed)
668 per_page: Items per page for page-based pagination
669 cursor: Cursor for cursor-based pagination
670 limit: Maximum items for cursor-based pagination (overrides default page size)
671 base_url: Base URL for link generation in page-based mode
672 query_params: Additional query parameters for links
674 Returns:
675 Union[Dict[str, Any], Tuple[List[Any], Optional[str]]]:
676 If page is provided: Dict with {"data": [...], "pagination": {...}, "links": {...}}
677 Otherwise: tuple of (list, next_cursor) for backward compatibility
679 Examples:
680 >>> import asyncio
681 >>> from unittest.mock import MagicMock
682 >>>
683 >>> # Test cursor-based pagination returns tuple format (list, next_cursor)
684 >>> async def test_cursor_based():
685 ... mock_db = MagicMock()
686 ... mock_query = MagicMock()
687 ... mock_query.column_descriptions = []
688 ... mock_query.limit = MagicMock(return_value=mock_query)
689 ... mock_db.execute = MagicMock(return_value=MagicMock(
690 ... scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
691 ... ))
692 ... result = await unified_paginate(
693 ... db=mock_db,
694 ... query=mock_query,
695 ... cursor=None,
696 ... limit=50
697 ... )
698 ... return isinstance(result, tuple) and len(result) == 2
699 >>> asyncio.run(test_cursor_based())
700 True
701 >>>
702 >>> # Verify return type difference: cursor mode returns tuple, page mode returns dict
703 >>> # Note: Page-based mode testing requires complex SQLAlchemy mocking,
704 >>> # see unit tests in tests/ for comprehensive page-based pagination tests
705 """
707 # Determine page size
708 if per_page is None:
709 per_page = limit if limit and limit > 0 else settings.pagination_default_page_size
711 # PAGE-BASED PAGINATION
712 if page is not None:
713 # Use existing paginate_query for page-based
714 result = await paginate_query(
715 db=db,
716 query=query,
717 page=page,
718 per_page=per_page,
719 base_url=base_url,
720 query_params=query_params,
721 use_cursor_threshold=False, # Explicit page-based mode
722 )
724 return result
726 # CURSOR-BASED PAGINATION
727 # Determine page size from limit parameter
728 if limit is not None:
729 if limit == 0:
730 page_size = None # No limit, fetch all
731 else:
732 page_size = min(limit, settings.pagination_max_page_size)
733 else:
734 page_size = settings.pagination_default_page_size
736 # Decode cursor if provided
737 # Standard
738 from datetime import datetime
740 last_id = None
741 last_created = None
742 if cursor:
743 try:
744 cursor_data = decode_cursor(cursor)
745 last_id = cursor_data.get("id")
746 created_str = cursor_data.get("created_at")
747 if created_str:
748 last_created = datetime.fromisoformat(created_str)
749 except (ValueError, TypeError) as e:
750 logger.warning(f"Invalid cursor, ignoring: {e}")
752 # Apply cursor filter with keyset pagination (assumes query already has ORDER BY)
753 if last_id and last_created:
754 # Extract model from query to apply filters
755 entities = query.column_descriptions
756 if entities:
757 # Third-Party
758 from sqlalchemy import and_, or_
760 model = entities[0]["entity"]
761 # Assumes descending order: created_at DESC, id DESC
762 query = query.where(or_(model.created_at < last_created, and_(model.created_at == last_created, model.id < last_id)))
764 # Fetch page_size + 1 to determine if there are more results
765 if page_size is not None:
766 query = query.limit(page_size + 1)
767 items = db.execute(query).scalars().all()
769 # Check if there are more results
770 has_more = False
771 if page_size is not None:
772 has_more = len(items) > page_size
773 if has_more:
774 items = items[:page_size]
776 # Generate next_cursor if there are more results
777 next_cursor = None
778 if has_more and items:
779 last_item = items[-1]
780 item_created_at = getattr(last_item, "created_at", None)
781 item_id = getattr(last_item, "id", None)
783 # Warn if cursor fields are missing - keyset pagination won't work correctly
784 if item_id is None:
785 logger.warning(
786 f"Cursor pagination may not work correctly: model {type(last_item).__name__} has no 'id' field. "
787 "Subsequent pages may return duplicate results. Consider using page-based pagination instead."
788 )
790 cursor_data = {"created_at": item_created_at, "id": item_id}
791 # Handle datetime serialization
792 if cursor_data["created_at"]:
793 cursor_data["created_at"] = cursor_data["created_at"].isoformat()
794 next_cursor = encode_cursor(cursor_data)
796 return (items, next_cursor)
799def parse_pagination_params(request: Request) -> Dict[str, Any]:
800 """Parse pagination parameters from request.
802 Args:
803 request: FastAPI request object
805 Returns:
806 Dictionary with parsed pagination parameters
808 Examples:
809 >>> from fastapi import Request
810 >>> # Mock request with query params
811 >>> request = type('Request', (), {
812 ... 'query_params': {'page': '2', 'per_page': '100'}
813 ... })()
814 >>> params = parse_pagination_params(request)
815 >>> params['page']
816 2
817 >>> params['per_page']
818 100
820 >>> # Test with cursor
821 >>> request_with_cursor = type('Request', (), {
822 ... 'query_params': {'cursor': 'eyJpZCI6IDEyM30=', 'per_page': '25'}
823 ... })()
824 >>> params_cursor = parse_pagination_params(request_with_cursor)
825 >>> params_cursor['cursor']
826 'eyJpZCI6IDEyM30='
827 >>> params_cursor['per_page']
828 25
830 >>> # Test with sort parameters
831 >>> request_with_sort = type('Request', (), {
832 ... 'query_params': {'page': '1', 'sort_by': 'name', 'sort_order': 'asc'}
833 ... })()
834 >>> params_sort = parse_pagination_params(request_with_sort)
835 >>> params_sort['sort_by']
836 'name'
837 >>> params_sort['sort_order']
838 'asc'
840 >>> # Test with invalid page (negative) - should default to 1
841 >>> request_invalid = type('Request', (), {
842 ... 'query_params': {'page': '-5', 'per_page': '50'}
843 ... })()
844 >>> params_invalid = parse_pagination_params(request_invalid)
845 >>> params_invalid['page']
846 1
848 >>> # Test with no parameters - uses defaults
849 >>> request_empty = type('Request', (), {'query_params': {}})()
850 >>> params_empty = parse_pagination_params(request_empty)
851 >>> params_empty['page']
852 1
853 >>> 'cursor' in params_empty
854 True
855 >>> 'sort_by' in params_empty
856 True
857 """
858 page = int(request.query_params.get("page", 1))
859 per_page = int(request.query_params.get("per_page", settings.pagination_default_page_size))
860 cursor = request.query_params.get("cursor")
861 sort_by = request.query_params.get("sort_by", settings.pagination_default_sort_field)
862 sort_order = request.query_params.get("sort_order", settings.pagination_default_sort_order)
864 # Validate and constrain values
865 page = max(1, page)
866 per_page = max(settings.pagination_min_page_size, min(per_page, settings.pagination_max_page_size))
868 return {
869 "page": page,
870 "per_page": per_page,
871 "cursor": cursor,
872 "sort_by": sort_by,
873 "sort_order": sort_order,
874 }