跳至内容

参考资料 hub_sdk/base/paginated_list.py

备注

该文件位于https://github.com/ultralytics/hub-sdk/blob/main/ hub_sdk/base/paginated_list .py。如果您发现问题,请通过提交 Pull Request🛠️ 帮助修复。谢谢🙏!



hub_sdk.base.paginated_list.PaginatedList

垒球 APIClient

源代码 hub_sdk/base/paginated_list.py
class PaginatedList(APIClient):
    def __init__(self, base_endpoint, name, page_size=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.pages = [None]
        self.current_page = 0
        self.total_pages = 1
        self._get()

    def _get(self, query=None):
        """
        Retrieve data for the current page.

        Args:
            query (dict, optional): Additional query parameters for the API request.
        """
        try:
            last_record = self.pages[self.current_page]
            resp = self.list(
                self.page_size,
                last_record,
                query=query,
            )
            self.__update_data(resp)
        except Exception as e:
            self.results = []
            self.logger.error("Failed to get data: %s", e)

    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("Failed to get previous page: %s", e)

    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("Failed to get next page: %s", e)

    def __update_data(self, resp: Response) -> None:
        """
        Update the internal data with the response from the API.

        Args:
            resp (Response): API response data.
        """
        resp_data = resp.json().get("data", {})
        self.results = resp_data.get("results", {})
        self.total_pages = resp_data.get("total") // self.page_size
        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

    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 = {"perPage": page_size}
            if last_record:
                params["lastRecordId"] = last_record
            if query:
                params["query"] = query
            return self.get("", params=params)
        except Exception as e:
            self.logger.error(f"Failed to list {self.name}: %s", e)

__init__(base_endpoint, name, page_size=None, headers=None)

初始化 PaginatedList 实例。

参数

名称 类型 说明 默认值
base_endpoint str

分页资源的基础 API 端点。

所需
name str

分页资源的描述性名称。

所需
page_size int

每页的项目数。

None
headers dict

在 API 请求中包含的附加标头。

None
源代码 hub_sdk/base/paginated_list.py
def __init__(self, base_endpoint, name, page_size=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.pages = [None]
    self.current_page = 0
    self.total_pages = 1
    self._get()

__update_data(resp)

根据 API 的响应更新内部数据。

参数

名称 类型 说明 默认值
resp Response

API 响应数据。

所需
源代码 hub_sdk/base/paginated_list.py
def __update_data(self, resp: Response) -> None:
    """
    Update the internal data with the response from the API.

    Args:
        resp (Response): API response data.
    """
    resp_data = resp.json().get("data", {})
    self.results = resp_data.get("results", {})
    self.total_pages = resp_data.get("total") // self.page_size
    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

list(page_size=10, last_record=None, query=None)

从 API 中读取项目列表。

参数

名称 类型 说明 默认值
page_size int

每页的项目数。

10
last_record str

上一页最后一条记录的 ID。

None
query dict

API 请求的附加查询参数。

None

返回:

类型 说明
Optional[Response]

列表请求的响应对象,如果失败则为 "无"。

源代码 hub_sdk/base/paginated_list.py
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 = {"perPage": page_size}
        if last_record:
            params["lastRecordId"] = last_record
        if query:
            params["query"] = query
        return self.get("", params=params)
    except Exception as e:
        self.logger.error(f"Failed to list {self.name}: %s", e)

next()

如果有结果,则转到下一页。

源代码 hub_sdk/base/paginated_list.py
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("Failed to get next page: %s", e)

previous()

如果有的话,移动到上一页结果。

源代码 hub_sdk/base/paginated_list.py
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("Failed to get previous page: %s", e)