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.

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
headers dict

Additional headers to include in API requests.

 None
Source code in hub_sdk/base/paginated_list.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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.
        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.

 required
Source code in hub_sdk/base/paginated_list.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __update_data(self, resp: Response) -> None:
    """
    Update the internal data with the response from the API.

    Args:
        resp (Response): API response data.
    """
    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.

 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 it fails.
Source code in hub_sdk/base/paginated_list.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, optional): The number of items per page.
        last_record (str, optional): ID of the last record from the previous page.
        query (dict, optional): Additional query parameters for the API request.

    Returns:
        (Optional[Response]): Response object from the list request, or None if it 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 
62
63
64
65
66
67
68
69
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 
53
54
55
56
57
58
59
60
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}")