Pagination

​

Learn how to use the pagination utilty in the SDK for efficient response handling

​

Key principles

1. Handle Large Datasets: Efficiently retrieve and process large amounts of data from APIs.
2. Optimize Performance: Reduce load on both the client and server by fetching data in smaller chunks.
3. Maintain Consistency: Provide a uniform interface for pagination across different API implementations.
4. Pagination Behaviour: Ensure that the page token returned is None when Lumos should stop calling the endpoint.

​

Implementing the pagination class

1. A Pagination class that represents the current page state.
2. A NextPageToken class that handles serialization and deserialization of pagination information or the above page state.

# Example: abc/pagination.py
from connector.utils.pagination import (
    NextPageTokenInterface,
    PaginationBase,
    create_next_page_token,
)

DEFAULT_PAGE_SIZE = 100

class AbcPagination(PaginationBase):
    resource_type: str | None = None
    page: int

    @classmethod
    def default(cls, resource_type: str) -> "AbcPagination":
        return cls(
            resource_type=resource_type,
            page=1,
        )

    def next_page(self) -> AbcPagination:
        return AbcPagination(
            resource_type=self.resource_type,
            page=self.page + 1,
        )

AbcNextPageToken = create_next_page_token(AbcPagination, "AbcNextPageToken")

def make_page(next: list[AbcPagination], size: int) -> Page | None:
    next_page_token = AbcNextPageToken.from_paginations(next)
    return Page(token=next_page_token.token, size=size) if next_page_token.token else None

1. PaginationClass: Represents the current page state (e.g., AbcPagination).
   • Includes any relevant fields like resource_type and page which you might need to paginate by.
     The resulting string token is using compression and can be populated with various parameters you might need to filter or paginate by on a per-capability basis. Note that all HTTP request parameters have size constraints.
   • Implements default method to create an initial pagination state.
   • Implements next_page method to generate the next page state.
2. NextPageTokenClass: Handles serialization and deserialization of pagination information.
   • Created using the create_next_page_token utility function.
   • Provides methods to convert between token strings and Pagination objects.
3. Helper functions, while not mandatory, are often added to:
   • Simplify the creation of Page objects (e.g., make_page function).
   • Handle specific pagination logic for the API
   • Provide utility methods for pagination-related operations

​

Using Pagination in Capabilities

# Example: abc/capabilities_read.py
from connector.oai.capability import get_page

def list_accounts(args: ListAccountsRequest) -> ListAccountsResponse:
	page = get_page(args)
	page_size = page.size or DEFAULT_PAGE_SIZE
	
	try:
	    current_pagination = NextPageToken(page.token).paginations()[0]
	except IndexError:
	    current_pagination = Pagination.default(resource_type)
	    
  async with AbcClient(args) as client:
    params = {
        "page": current_pagination.page,
        "perPage": page_size,
    }
    response = await client.getUsers(params=params)
    data = response.json()
    accounts = data.to_accounts()
    
    if more_data_available:
		    next_pagination = [current_pagination.next_page()]
		else:
		    next_pagination = []
		
		next_page = make_page(next_pagination, page_size)
		
		return ListAccountsResponse(
				response=accounts,
				page=next_page,
		)

​

Key considerations

1. The get_page(args) helper function is used to retrieve pagination information from the request arguments.
2. page_size is determined using the page size from the request or a default value.
3. The same page_size is used across all pages to maintain consistency.
4. Instead of using template strings for endpoints, the endpoint is typically stored in the pagination object or determined based on the resource type or custom argument.

​

Handling Multiple Endpoint Pagination

# Example: abc/capabilities_read.py:list_accounts
# ...
page = get_page(args)
page_size = page.size or DEFAULT_PAGE_SIZE

try:
    current_paginations = NextPageToken(page.token).paginations()
except IndexError:
    current_paginations = [Pagination.default(endpoint="/first-endpoint")]

async with Abcclient as client:
    for pagination in current_paginations:
        params = {
            "page": pagination.page,
            "perPage": page_size,
        }
        response = await client.get(pagination.endpoint, params=params)
        # ...

​

Storing Additional Information in the Pagination Tokens

class AbcPagination(PaginationBase):
    page: int
    unique_ids: list[str]

    @classmethod
    def default(cls, endpoint: str) -> "AbcPagination":
        return cls(endpoint=endpoint, page=1, unique_ids=list())

​

Best practices

1. Store Minimal Data: Only store essential information in the pagination token. In the example, only unique identifiers are stored, not entire objects.
2. Use Efficient Data Structures: Choose data structures that balance functionality and size. For example, using a list of strings for unique IDs instead of more complex objects.
3. Limit Data Volume: If the amount of data grows too large, consider implementing a cutoff or using a more efficient storage method.