pagination.py•6.44 kB
"""pagination/streaming detection and adapter."""
from __future__ import annotations
import inspect
from typing import Any, Iterator, List, Optional, Dict
from dataclasses import dataclass
@dataclass
class PaginationConfig:
"""detected pagination configuration for a tool."""
has_pagination: bool
cursor_param: Optional[str] = None # e.g., "page", "next_token", "offset"
limit_param: Optional[str] = None # e.g., "per_page", "limit", "page_size"
is_iterator: bool = False
max_items_default: int = 100
class PaginationDetector:
"""detects pagination patterns in sdk methods."""
# common pagination parameter names
CURSOR_PARAMS = [
"page", "next_token", "continuation_token", "offset",
"cursor", "marker", "bookmark", "starting_after"
]
LIMIT_PARAMS = [
"per_page", "limit", "page_size", "max_results",
"count", "size", "top"
]
def detect(self, callable_obj: Any, signature: Optional[inspect.Signature]) -> PaginationConfig:
"""detect if a callable supports pagination."""
if not signature:
return PaginationConfig(has_pagination=False)
# check if return type is iterator/generator
is_iterator = self._is_iterator_return(callable_obj)
# check for pagination parameters
cursor_param = None
limit_param = None
for param_name in signature.parameters:
param_lower = param_name.lower()
if not cursor_param:
for pattern in self.CURSOR_PARAMS:
if pattern in param_lower:
cursor_param = param_name
break
if not limit_param:
for pattern in self.LIMIT_PARAMS:
if pattern in param_lower:
limit_param = param_name
break
has_pagination = is_iterator or cursor_param is not None or limit_param is not None
return PaginationConfig(
has_pagination=has_pagination,
cursor_param=cursor_param,
limit_param=limit_param,
is_iterator=is_iterator
)
def _is_iterator_return(self, callable_obj: Any) -> bool:
"""check if return type is an iterator."""
try:
hints = inspect.get_annotations(callable_obj)
if "return" in hints:
ret_type = hints["return"]
ret_str = str(ret_type).lower()
return any(pattern in ret_str for pattern in [
"iterator", "generator", "iterable", "pager"
])
except Exception:
pass
return False
class PaginationAdapter:
"""wraps paginated calls to handle auto-iteration."""
def __init__(self, max_items: int = 100):
"""
initialize pagination adapter.
args:
max_items: maximum items to fetch across pages
"""
self.max_items = max_items
def execute(
self,
callable_obj: Any,
args: tuple,
kwargs: dict,
config: PaginationConfig
) -> Any:
"""execute a potentially paginated call."""
if not config.has_pagination:
return callable_obj(*args, **kwargs)
# check if user provided explicit pagination controls
if config.limit_param and config.limit_param in kwargs:
# user wants specific page size, don't auto-iterate
return callable_obj(*args, **kwargs)
# auto-iterate if it's an iterator
if config.is_iterator:
return self._collect_from_iterator(callable_obj(*args, **kwargs))
# if has cursor/limit params, set sensible defaults
if config.limit_param and config.limit_param not in kwargs:
kwargs[config.limit_param] = min(self.max_items, 50) # reasonable page size
result = callable_obj(*args, **kwargs)
# check if result looks like a paginated response
if self._is_paginated_result(result):
return self._extract_items(result)
return result
def _collect_from_iterator(self, iterator: Iterator) -> List[Any]:
"""collect items from an iterator up to max_items."""
items = []
try:
for item in iterator:
items.append(item)
if len(items) >= self.max_items:
break
except Exception:
# iterator exhausted or error
pass
return items
def _is_paginated_result(self, result: Any) -> bool:
"""check if result is a paginated response object."""
if not hasattr(result, "__dict__"):
return False
attrs = dir(result)
# common pagination result patterns
return any(attr in attrs for attr in [
"items", "results", "data", "values",
"next_page_token", "next_token", "has_more"
])
def _extract_items(self, result: Any) -> Dict[str, Any]:
"""extract items and pagination metadata from result."""
response: Dict[str, Any] = {}
# try to find the items list
for attr in ["items", "results", "data", "values"]:
if hasattr(result, attr):
items = getattr(result, attr)
if isinstance(items, list):
response["items"] = [self._serialize_item(i) for i in items[:self.max_items]]
break
# extract pagination metadata
for attr in ["next_page_token", "next_token", "continuation_token", "next_cursor"]:
if hasattr(result, attr):
response["next_cursor"] = getattr(result, attr)
break
if hasattr(result, "has_more"):
response["has_more"] = getattr(result, "has_more")
return response if response else result
def _serialize_item(self, item: Any) -> Any:
"""serialize an item to json-compatible format."""
if hasattr(item, "to_dict"):
return item.to_dict()
elif hasattr(item, "__dict__"):
return {k: v for k, v in item.__dict__.items() if not k.startswith("_")}
else:
return str(item)