tessl/pypi-msrest

AutoRest swagger generator Python client runtime for REST API clients with serialization, authentication, and request handling.

—

Pending

Overview

Eval results

Files

Paging

Name: tessl/pypi-msrest
Author: tessl

Iterator-based paging support for REST APIs that return large result sets, providing automatic page traversal, reset capability, and raw response access. The paging system enables efficient handling of paginated API responses with Python iterator protocol.

Capabilities

Paged Iterator

Main class for handling paginated REST responses with automatic page traversal.

class Paged:
    def __init__(self, command, classes, raw_headers=None, **kwargs):
        """
        Initialize paged response container.
        
        Parameters:
        - command: Function to retrieve next page (callable)
        - classes: Dict of model classes for deserialization
        - raw_headers: Dict of headers to include in raw responses
        - kwargs: Additional initialization parameters
        """
    
    next_link: str = ""
    current_page: list = []
    
    def __iter__(self):
        """Return iterator (self)."""
    
    def __next__(self):
        """Get next item, advancing pages as needed."""
    
    # Python 2 compatibility
    next = __next__

Page Navigation

Control page traversal and iteration state.

def advance_page(self) -> list:
    """
    Force moving cursor to next page.
    
    Returns:
    Current page list
    
    Raises:
    StopIteration if no further pages available
    """

def reset(self):
    """Reset iterator to first page."""

def get(self, url: str) -> list:
    """
    Get arbitrary page by URL.
    
    This resets iterator and consumes it to return specific page.
    
    Parameters:
    - url: URL to specific page
    
    Returns:
    Page items list
    """

Raw Response Access

Access underlying HTTP response data and headers.

@property
def raw(self):
    """
    Get current page as ClientRawResponse.
    
    Returns:
    ClientRawResponse with current page data and response headers
    """

Async Support

For Python 3.5+, async paging support is available through AsyncPagedMixin.

class AsyncPagedMixin:
    """Mixin providing async iterator support for paging."""
    
    async def __aiter__(self):
        """Return async iterator."""
    
    async def __anext__(self):
        """Get next item asynchronously."""

Usage Examples

Basic Paging

from msrest.paging import Paged
from msrest import ServiceClient, Configuration

# Assuming you have a function that gets the next page
def get_next_page(next_link):
    """Function to fetch next page from API."""
    if not next_link:
        # First page
        request = client.get('/users')
    else:
        # Subsequent pages
        request = client.get(next_link)
    
    response = client.send(request)
    return response

# Create configuration and client
config = Configuration(base_url='https://api.example.com')
client = ServiceClient(None, config)

# Create paged iterator
model_classes = {'User': User}  # Your model classes
users_paged = Paged(get_next_page, model_classes)

# Iterate through all pages automatically
for user in users_paged:
    print(f"User: {user.name} ({user.email})")

Manual Page Control

from msrest.paging import Paged

# Create paged iterator
users_paged = Paged(get_next_page, model_classes)

# Get first page manually
first_page = users_paged.advance_page()
print(f"First page has {len(first_page)} users")

# Check if more pages available
if users_paged.next_link:
    second_page = users_paged.advance_page()
    print(f"Second page has {len(second_page)} users")

# Reset to beginning
users_paged.reset()

# Start iteration from beginning again
for user in users_paged:
    print(user.name)
    break  # Just get first user

Direct Page Access

from msrest.paging import Paged

users_paged = Paged(get_next_page, model_classes)

# Get specific page by URL
specific_page_url = 'https://api.example.com/users?page=5'
page_5_users = users_paged.get(specific_page_url)

print(f"Page 5 has {len(page_5_users)} users")
for user in page_5_users:
    print(f"  - {user.name}")

Raw Response Access

from msrest.paging import Paged

# Include headers in raw responses
headers_to_capture = {'X-Total-Count': 'str', 'X-Page-Count': 'str'}
users_paged = Paged(get_next_page, model_classes, raw_headers=headers_to_capture)

# Get first page
users_paged.advance_page()

# Access raw response data
raw_response = users_paged.raw
print(f"Response status: {raw_response.response.status_code}")
print(f"Total count: {raw_response.headers.get('X-Total-Count')}")
print(f"Page count: {raw_response.headers.get('X-Page-Count')}")
print(f"Current page items: {len(raw_response.output)}")

Custom Page Retrieval Function

import json
from msrest.paging import Paged

def custom_page_fetcher(next_link):
    """Custom function to handle specific API pagination format."""
    
    if not next_link:
        # First request
        url = '/api/data'
        params = {'page': 1, 'per_page': 50}
    else:
        # Parse next_link to get page info
        from urllib.parse import urlparse, parse_qs
        parsed = urlparse(next_link)
        query_params = parse_qs(parsed.query)
        
        url = parsed.path
        params = {k: v[0] for k, v in query_params.items()}
    
    # Make request
    request = client.get(url, params=params)
    response = client.send(request)
    
    # Parse response to extract items and next link
    data = json.loads(response.text)
    
    # Assuming API returns: {"items": [...], "next_page": "url"}
    # Transform to msrest format
    response_with_items = type('Response', (), {
        'text': json.dumps(data['items']),
        'headers': response.headers,
        'status_code': response.status_code
    })()
    
    # Set next link for paging
    if 'next_page' in data:
        response_with_items.next_link = data['next_page']
    else:
        response_with_items.next_link = None
    
    return response_with_items

# Use custom fetcher
data_paged = Paged(custom_page_fetcher, {'DataItem': DataItem})

# Iterate through all pages
for item in data_paged:
    print(f"Item: {item.id}")

Async Paging (Python 3.5+)

import asyncio
from msrest.paging import Paged

async def async_page_fetcher(next_link):
    """Async function to fetch pages."""
    # Use async HTTP client here
    # This is a simplified example
    
    if not next_link:
        url = '/async/data'
    else:
        url = next_link
    
    # Async request (pseudo-code)
    async_response = await async_client.get(url)
    return async_response

# For Python 3.5+
class AsyncPaged(Paged):
    """Async version of Paged iterator."""
    
    async def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current_page and self._current_page_iter_index < len(self.current_page):
            response = self.current_page[self._current_page_iter_index]
            self._current_page_iter_index += 1
            return response
        else:
            await self.advance_page_async()
            return await self.__anext__()
    
    async def advance_page_async(self):
        if self.next_link is None:
            raise StopAsyncIteration("End of paging")
        
        self._current_page_iter_index = 0
        self._response = await self._get_next(self.next_link)
        self._derserializer(self, self._response)
        return self.current_page

# Usage
async def process_async_pages():
    async_paged = AsyncPaged(async_page_fetcher, model_classes)
    
    async for item in async_paged:
        print(f"Async item: {item.name}")
        
        # Process only first 10 items
        if some_condition:
            break

# Run async function
asyncio.run(process_async_pages())

Error Handling with Paging

from msrest.paging import Paged
from msrest.exceptions import HttpOperationError

def robust_page_fetcher(next_link):
    """Page fetcher with error handling."""
    try:
        if not next_link:
            request = client.get('/data')
        else:
            request = client.get(next_link)
        
        response = client.send(request)
        return response
        
    except HttpOperationError as e:
        if e.response.status_code == 404:
            # No more pages
            return None
        else:
            # Re-raise other errors
            raise

data_paged = Paged(robust_page_fetcher, model_classes)

try:
    for item in data_paged:
        print(f"Item: {item.id}")
        
except StopIteration:
    print("Finished processing all pages")
except HttpOperationError as e:
    print(f"API error during paging: {e}")

Paging with Authentication

from msrest import ServiceClient, Configuration
from msrest.authentication import ApiKeyCredentials
from msrest.paging import Paged

# Setup authenticated client
config = Configuration(base_url='https://api.example.com')
config.credentials = ApiKeyCredentials(in_headers={'Authorization': 'Bearer your-token'})

with ServiceClient(None, config) as client:
    def auth_page_fetcher(next_link):
        """Authenticated page fetcher."""
        if not next_link:
            request = client.get('/protected/data')
        else:
            request = client.get(next_link)
        
        # Authentication is automatically handled by the client
        return client.send(request)
    
    # Create paged iterator with authenticated fetcher
    protected_data = Paged(auth_page_fetcher, model_classes)
    
    # Iterate through protected resources
    for item in protected_data:
        print(f"Protected item: {item.name}")

Integration with Deserializer

The Paged class works closely with the Deserializer to convert raw response data into Python objects:

from msrest.serialization import Deserializer
from msrest.paging import Paged

# The Paged class internally uses a deserializer
# You provide the model classes during initialization
model_classes = {
    'User': User,
    'Address': Address,
    'Organization': Organization
}

# Paged creates internal deserializer: Deserializer(model_classes)
users_paged = Paged(get_next_page, model_classes)

# Each page response is automatically deserialized using the appropriate model class
for user in users_paged:
    # user is already a User model instance
    print(f"User: {user.name}")
    if hasattr(user, 'address'):
        print(f"  Address: {user.address.city}")

Install with Tessl CLI