CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-google-api-core

Google API client core library providing common helpers, utilities, and components for Python client libraries

—

Pending
Overview
Eval results
Files

page-iteration.mddocs/

Page Iteration

Standardized pagination patterns for list operations with support for both HTTP/JSON and gRPC APIs, including async variants. The page iteration system provides consistent interfaces for traversing large result sets across different Google API transport protocols.

Capabilities

Base Iterator Classes

Abstract base classes that define the common interface for all page iterators.

class Iterator:
    """
    Abstract base class for iterating through API responses.
    
    Args:
        client: API client instance
        item_to_value (Callable): Function to transform items from API responses
        page_token (str, optional): Token for starting at specific page
        max_results (int, optional): Maximum number of results to return
    """
    def __init__(self, client, item_to_value, page_token=None, max_results=None): ...
    
    def __iter__(self):
        """Return iterator instance."""
        return self
    
    def __next__(self):
        """
        Get next item from iterator.
        
        Returns:
            Any: Next item from the API response
            
        Raises:
            StopIteration: When no more items available
        """
    
    @property
    def page_number(self):
        """
        Current page number (0-indexed).
        
        Returns:
            int: Current page number
        """
    
    @property
    def next_page_token(self):
        """
        Token for the next page of results.
        
        Returns:
            str or None: Next page token, None if no more pages
        """
    
    @property 
    def num_results(self):
        """
        Total number of results yielded so far.
        
        Returns:
            int: Count of results returned
        """

class AsyncIterator:
    """
    Abstract base class for async iteration through API responses.
    
    Args:
        client: Async API client instance
        item_to_value (Callable): Function to transform items from API responses
        page_token (str, optional): Token for starting at specific page
        max_results (int, optional): Maximum number of results to return
    """
    def __init__(self, client, item_to_value, page_token=None, max_results=None): ...
    
    def __aiter__(self):
        """Return async iterator instance."""
        return self
    
    async def __anext__(self):
        """
        Get next item from async iterator.
        
        Returns:
            Any: Next item from the API response
            
        Raises:
            StopAsyncIteration: When no more items available
        """
    
    @property
    def page_number(self):
        """Current page number (0-indexed)."""
    
    @property 
    def next_page_token(self):
        """Token for the next page of results."""
    
    @property
    def num_results(self):
        """Total number of results yielded so far."""

HTTP/REST Iterator Classes

Iterators for HTTP/JSON-based APIs with RESTful pagination patterns.

class HTTPIterator(Iterator):
    """
    Iterator for HTTP/JSON API list responses.
    
    Args:
        client: HTTP API client instance
        api_request (Callable): Function to make HTTP API requests
        path (str): API endpoint path template
        item_to_value (Callable): Function to extract items from response
        page_token (str, optional): Starting page token
        max_results (int, optional): Maximum results to return
        extra_params (dict, optional): Additional query parameters
    """
    def __init__(self, client, api_request, path, item_to_value, page_token=None, max_results=None, extra_params=None): ...
    
    @property
    def started(self):
        """
        Check if iteration has started.
        
        Returns:
            bool: True if iteration has begun
        """
    
    def _update_state(self, response):
        """
        Update iterator state from API response.
        
        Args:
            response (dict): API response data
        """

gRPC Iterator Classes

Iterators for gRPC-based APIs with protobuf message pagination.

class GRPCIterator(Iterator):
    """
    Iterator for gRPC list responses using protobuf messages.
    
    Args:
        client: gRPC client instance
        method (Callable): gRPC method to call for pagination
        request: Initial request protobuf message
        items_field (str): Field name containing items in response
        request_token_field (str): Field name for page token in request (default: "page_token")
        response_token_field (str): Field name for next page token in response (default: "next_page_token")
    """
    def __init__(self, client, method, request, items_field, request_token_field="page_token", response_token_field="next_page_token"): ...
    
    @property
    def started(self):
        """Check if iteration has started."""
        
    def _update_state(self, response):
        """Update iterator state from gRPC response."""

class AsyncGRPCIterator(AsyncIterator):
    """
    Async iterator for gRPC list responses.
    
    Args:
        client: Async gRPC client instance
        method (Callable): Async gRPC method for pagination
        request: Initial request protobuf message
        items_field (str): Field name containing items in response
        request_token_field (str): Page token field in request
        response_token_field (str): Next page token field in response
    """
    def __init__(self, client, method, request, items_field, request_token_field="page_token", response_token_field="next_page_token"): ...
    
    @property
    def started(self):
        """Check if async iteration has started."""
    
    async def _update_state(self, response):
        """Update iterator state from async gRPC response."""

Page Container Classes

Classes representing individual pages of results within an iterator.

class Page:
    """
    Single page of results from a paginated API response.
    
    Args:
        parent (Iterator): Parent iterator instance
        response: Raw API response for this page
        item_to_value (Callable): Function to transform response items
    """
    def __init__(self, parent, response, item_to_value): ...
    
    def __iter__(self):
        """Iterate over items in this page."""
    
    @property
    def num_items(self):
        """
        Number of items in this page.
        
        Returns:
            int: Count of items in current page
        """
    
    @property
    def remaining(self):
        """
        Number of items remaining in this page.
        
        Returns:
            int: Count of unprocessed items in page
        """
    
    @property
    def response(self):
        """
        Raw API response for this page.
        
        Returns:
            Any: Original API response object
        """

Utility Functions

Helper functions for customizing page iteration behavior.

def _item_to_value_identity(iterator, item):
    """
    Default item transformation function that returns items unchanged.
    
    Args:
        iterator: Iterator instance (unused)
        item: Item from API response
        
    Returns:
        Any: Item unchanged
    """

def _do_nothing_page_start(iterator, page, response):
    """
    Default page start handler that performs no action.
    
    Args:
        iterator: Iterator instance
        page: Page instance 
        response: API response for the page
    """

Usage Examples

Basic HTTP API Pagination

from google.api_core import page_iterator
import requests

class HTTPClient:
    def __init__(self, base_url):
        self.base_url = base_url
    
    def api_request(self, method, path, **kwargs):
        """Make HTTP API request."""
        url = f"{self.base_url}{path}"
        response = requests.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()

def extract_items(response):
    """Extract items from API response."""
    return response.get("items", [])

# Create HTTP iterator
client = HTTPClient("https://api.example.com")
iterator = page_iterator.HTTPIterator(
    client=client,
    api_request=lambda path, **kwargs: client.api_request("GET", path, params=kwargs),
    path="/users",
    item_to_value=lambda iterator, item: item,
    max_results=100
)

# Iterate through all results
users = []
for user in iterator:
    users.append(user)
    print(f"User: {user['name']}")

print(f"Total users retrieved: {len(users)}")
print(f"Pages processed: {iterator.page_number + 1}")

gRPC API Pagination

from google.api_core import page_iterator
import grpc
from my_api_pb2 import ListUsersRequest
from my_api_pb2_grpc import UserServiceStub

# Create gRPC client
channel = grpc.insecure_channel("api.example.com:443")
client = UserServiceStub(channel)

# Create initial request
request = ListUsersRequest(
    page_size=50,
    filter="status:active"
)

# Create gRPC iterator
iterator = page_iterator.GRPCIterator(
    client=client,
    method=client.ListUsers,
    request=request,
    items_field="users"  # Field containing user list in response
)

# Iterate through users
for user in iterator:
    print(f"User ID: {user.id}, Name: {user.name}")
    
print(f"Total results: {iterator.num_results}")

Async Pagination

import asyncio
from google.api_core import page_iterator_async
import aiohttp

class AsyncHTTPClient:
    def __init__(self, base_url):
        self.base_url = base_url
    
    async def api_request(self, path, **params):
        """Make async HTTP API request."""
        async with aiohttp.ClientSession() as session:
            url = f"{self.base_url}{path}"
            async with session.get(url, params=params) as response:
                response.raise_for_status()
                return await response.json()

async def async_pagination_example():
    client = AsyncHTTPClient("https://api.example.com")
    
    # Create async iterator
    iterator = page_iterator_async.AsyncHTTPIterator(
        client=client,
        api_request=client.api_request,
        path="/products", 
        item_to_value=lambda iterator, item: item,
        max_results=200
    )
    
    # Async iteration
    products = []
    async for product in iterator:
        products.append(product)
        print(f"Product: {product['name']}")
    
    print(f"Total products: {len(products)}")

asyncio.run(async_pagination_example())

Custom Item Transformation

from google.api_core import page_iterator
from datetime import datetime

def transform_user_item(iterator, raw_user):
    """Transform raw API user data into custom format."""
    return {
        "id": raw_user["user_id"],
        "name": raw_user["display_name"],
        "email": raw_user["email_address"],
        "created": datetime.fromisoformat(raw_user["created_at"]),
        "active": raw_user["status"] == "active"
    }

# Use custom transformation
iterator = page_iterator.HTTPIterator(
    client=client,
    api_request=api_request_func,
    path="/users",
    item_to_value=transform_user_item,
    max_results=1000
)

# Get transformed user objects
for user in iterator:
    print(f"User {user['name']} ({user['email']}) - Active: {user['active']}")

Page-by-Page Processing

from google.api_core import page_iterator

# Create iterator
iterator = page_iterator.HTTPIterator(
    client=client,
    api_request=api_request_func,
    path="/large-dataset",
    item_to_value=lambda iterator, item: item
)

# Process one page at a time
for page in iterator._page_iter:
    print(f"Processing page {iterator.page_number + 1}")
    print(f"Page contains {page.num_items} items")
    
    # Process all items in current page
    page_items = list(page)
    
    # Custom page processing logic
    process_page_batch(page_items)
    
    print(f"Completed page {iterator.page_number + 1}")

Iterator with Error Handling

from google.api_core import page_iterator
from google.api_core import exceptions
from google.api_core import retry

# Configure retry for API requests
retry_config = retry.Retry(
    predicate=retry.if_exception_type(
        exceptions.ServiceUnavailable,
        exceptions.InternalServerError
    ),
    initial=1.0,
    maximum=60.0
)

@retry_config
def robust_api_request(path, **params):
    """API request with retry logic."""
    response = requests.get(f"https://api.example.com{path}", params=params)
    if response.status_code >= 500:
        raise exceptions.ServiceUnavailable("Service unavailable")
    response.raise_for_status()
    return response.json()

# Create iterator with robust error handling
iterator = page_iterator.HTTPIterator(
    client=client,
    api_request=robust_api_request,
    path="/reliable-endpoint",
    item_to_value=lambda iterator, item: item
)

# Iterate with error handling
try:
    items = list(iterator)
    print(f"Successfully retrieved {len(items)} items")
except exceptions.GoogleAPIError as e:
    print(f"API error during pagination: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Install with Tessl CLI

npx tessl i tessl/pypi-google-api-core

docs

bidirectional-streaming.md

client-config.md

datetime.md

exceptions.md

gapic-framework.md

iam-policies.md

index.md

operations.md

page-iteration.md

path-templates.md

protobuf-helpers.md

retry.md

timeout.md

transport.md

universe-domain.md

tile.json