์ฝ˜ํ…์ธ ๋กœ ๊ฑด๋„ˆ๋›ฐ๊ธฐ

์ฐธ์กฐ hub_sdk/base/paginated_list.py

์ฐธ๊ณ 

์ด ํŒŒ์ผ์€ https://github.com/ultralytics/hub-sdk/blob/main/hub_sdk/base/paginated_list .py์—์„œ ํ™•์ธํ•  ์ˆ˜ ์žˆ์Šต๋‹ˆ๋‹ค. ๋ฌธ์ œ๋ฅผ ๋ฐœ๊ฒฌํ•˜๋ฉด ํ’€ ๋ฆฌํ€˜์ŠคํŠธ ๐Ÿ› ๏ธ ์— ๊ธฐ์—ฌํ•˜์—ฌ ๋ฌธ์ œ๋ฅผ ํ•ด๊ฒฐํ•˜๋„๋ก ๋„์™€์ฃผ์„ธ์š”. ๊ฐ์‚ฌํ•ฉ๋‹ˆ๋‹ค ๐Ÿ™!



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, 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()

    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(f"Failed to get data: {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(f"Failed to get previous page: {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(f"Failed to get next page: {e}")

    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)

    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}")

__init__(base_endpoint, name, page_size=None, public=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, 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(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.
    """
    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(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 = {"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()

๊ฐ€๋Šฅํ•œ ๊ฒฝ์šฐ ๋‹ค์Œ ๊ฒฐ๊ณผ ํŽ˜์ด์ง€๋กœ ์ด๋™ํ•ฉ๋‹ˆ๋‹ค.

์˜ ์†Œ์Šค ์ฝ”๋“œ 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(f"Failed to get next page: {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(f"Failed to get previous page: {e}")