Skip to content

Reference for hub_sdk/base/paginated_list.py

Note

This file is available at https://github.com/ultralytics/hub-sdk/blob/main/hub_sdk/base/paginated_list.py. If you spot a problem please help fix it by contributing a Pull Request 🛠️. Thank you 🙏!


hub_sdk.base.paginated_list.PaginatedList 

PaginatedList(base_endpoint, name, page_size=None, public=None, headers=None)

Bases: APIClient

Handles pagination for list endpoints on the API while managing retrieval, navigation, and updating of data.

This class extends APIClient to provide pagination functionality for API endpoints that return large datasets. It manages page navigation, data retrieval, and state tracking across paginated results.

Attributes:

Name Type Description
name str

Descriptive name for the paginated resource.
page_size int

Number of items to display per page.
public bool

Filter for public resources if specified.
pages List

List tracking page identifiers for navigation.
current_page int

Index of the currently displayed page.
total_pages int

Total number of available pages.
results dict

Current page results from the API.

Methods:

Name Description
previous

Navigate to the previous page of results.
next

Navigate to the next page of results.
list

Retrieve a list of items from the API with pagination parameters.

Parameters:

Name Type Description Default
base_endpoint str

The base API endpoint for the paginated resource.

 required
name str

A descriptive name for the paginated resource.

 required
page_size int

The number of items per page.

 None
public bool

Filter for public resources if specified.

 None
headers dict

Additional headers to include in API requests.

 None
Source code in hub_sdk/base/paginated_list.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(self, base_endpoint, name, page_size=None, public=None, headers=None):
    """
    Initialize a PaginatedList instance.

    Args:
        base_endpoint (str): The base API endpoint for the paginated resource.
        name (str): A descriptive name for the paginated resource.
        page_size (int, optional): The number of items per page.
        public (bool, optional): Filter for public resources if specified.
        headers (dict, optional): Additional headers to include in API requests.
    """
    super().__init__(f"{HUB_FUNCTIONS_ROOT}/v1/{base_endpoint}", headers)
    self.name = name
    self.page_size = page_size
    self.public = public
    self.pages = [None]
    self.current_page = 0
    self.total_pages = 1
    self._get()

__update_data 

__update_data(resp: Response) -> None

Update the internal data with the response from the API.

Parameters:

Name Type Description Default
resp Response

API response data containing pagination information and results.

 required
Source code in hub_sdk/base/paginated_list.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def __update_data(self, resp: Response) -> None:
    """
    Update the internal data with the response from the API.

    Args:
        resp (Response): API response data containing pagination information and results.
    """
    if resp:
        resp_data = resp.json().get("data", {})
        self.results = resp_data.get("results", {})
        self.total_pages = math.ceil(resp_data.get("total") / self.page_size) if self.page_size > 0 else 0
        last_record_id = resp_data.get("lastRecordId")
        if last_record_id is None:
            self.pages[self.current_page + 1 :] = [None] * (len(self.pages) - self.current_page - 1)
        elif len(self.pages) <= self.current_page + 1:
            self.pages.append(last_record_id)
        else:
            self.pages[self.current_page + 1] = last_record_id
    else:
        self.results = {}
        self.total_pages = 0
        self.pages[self.current_page + 1 :] = [None] * (len(self.pages) - self.current_page - 1)

list 

list(page_size: int = 10, last_record=None, query=None) -> Optional[Response]

Retrieve a list of items from the API.

Parameters:

Name Type Description Default
page_size int

The number of items per page.

 10
last_record str

ID of the last record from the previous page for cursor-based pagination.

 None
query dict

Additional query parameters for the API request.

 None

Returns:

Type Description
Optional[Response]

Response object from the list request, or None if the request fails.
Source code in hub_sdk/base/paginated_list.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def list(self, page_size: int = 10, last_record=None, query=None) -> Optional[Response]:
    """
    Retrieve a list of items from the API.

    Args:
        page_size (int): The number of items per page.
        last_record (str, optional): ID of the last record from the previous page for cursor-based pagination.
        query (dict, optional): Additional query parameters for the API request.

    Returns:
        (Optional[Response]): Response object from the list request, or None if the request fails.
    """
    try:
        params = {"limit": page_size}
        if last_record:
            params["last_doc_id"] = last_record
        if query:
            params["query"] = query
        if self.public is not None:
            params["public"] = self.public
        return self.get("", params=params)
    except Exception as e:
        self.logger.error(f"Failed to list {self.name}: {e}")

next 

next() -> None

Move to the next page of results if available.

Source code in hub_sdk/base/paginated_list.py 
82
83
84
85
86
87
88
89
def next(self) -> None:
    """Move to the next page of results if available."""
    try:
        if self.current_page < self.total_pages - 1:
            self.current_page += 1
            self._get()
    except Exception as e:
        self.logger.error(f"Failed to get next page: {e}")

previous 

previous() -> None

Move to the previous page of results if available.

Source code in hub_sdk/base/paginated_list.py 
73
74
75
76
77
78
79
80
def previous(self) -> None:
    """Move to the previous page of results if available."""
    try:
        if self.current_page > 0:
            self.current_page -= 1
            self._get()
    except Exception as e:
        self.logger.error(f"Failed to get previous page: {e}")



📅 Created 1 year ago ✏️ Updated 1 month ago
glenn-jocher