Skip to content

Référence pour hub_sdk/base/paginated_list.py

Note

Ce fichier est disponible à l'adresse https://github.com/ultralytics/ hub-sdk/blob/main/ hub_sdk/base/paginated_list .py. Si tu repères un problème, aide à le corriger en contribuant à une Pull Request 🛠️. Merci 🙏 !



hub_sdk.base.paginated_list.PaginatedList

Bases : APIClient

Code source dans 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)

Initialise une instance de PaginatedList.

Paramètres :

Nom Type Description DĂ©faut
base_endpoint str

Le point de terminaison de l'API de base pour la ressource paginée.

requis
name str

Un nom descriptif pour la ressource paginée.

requis
page_size int

Le nombre d'éléments par page.

None
headers dict

En-têtes supplémentaires à inclure dans les demandes d'API.

None
Code source dans 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)

Mets à jour les données internes avec la réponse de l'API.

Paramètres :

Nom Type Description DĂ©faut
resp Response

Données de la réponse de l'API.

requis
Code source dans 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)

Récupère une liste d'éléments de l'API.

Paramètres :

Nom Type Description DĂ©faut
page_size int

Le nombre d'éléments par page.

10
last_record str

ID du dernier enregistrement de la page précédente.

None
query dict

Paramètres de requête supplémentaires pour la demande d'API.

None

Retourne :

Type Description
Optional[Response]

Objet de réponse de la demande de liste, ou Aucun en cas d'échec.

Code source dans 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()

Passe à la page suivante de résultats si elle est disponible.

Code source dans 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()

Passe à la page précédente des résultats si elle est disponible.

Code source dans 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)