############################ Copyrights and license ############################ # # # Copyright 2012 Vincent Jacques # # Copyright 2012 Zearin # # Copyright 2013 AKFish # # Copyright 2013 Bill Mill # # Copyright 2013 Vincent Jacques # # Copyright 2013 davidbrai # # Copyright 2014 Thialfihar # # Copyright 2014 Vincent Jacques # # Copyright 2015 Dan Vanderkam # # Copyright 2015 Eliot Walker # # Copyright 2016 Peter Buckley # # Copyright 2017 Jannis Gebauer # # Copyright 2018 Gilad Shefer # # Copyright 2018 Joel Koglin # # Copyright 2018 Steve Kowalik # # Copyright 2018 Wan Liuyang # # Copyright 2018 netsgnut <284779+netsgnut@users.noreply.github.com> # # Copyright 2018 sfdye # # Copyright 2019 Steve Kowalik # # Copyright 2019 Wan Liuyang # # Copyright 2020 Emir Hodzic # # Copyright 2020 Steve Kowalik # # Copyright 2021 Mark Walker # # Copyright 2021 Steve Kowalik # # Copyright 2023 Andrew Dawes <53574062+AndrewJDawes@users.noreply.github.com> # # Copyright 2023 Enrico Minack # # Copyright 2023 Jirka Borovec <6035284+Borda@users.noreply.github.com> # # Copyright 2023 Trim21 # # Copyright 2023 YugoHino # # Copyright 2024 Enrico Minack # # # # This file is part of PyGithub. # # http://pygithub.readthedocs.io/ # # # # PyGithub is free software: you can redistribute it and/or modify it under # # the terms of the GNU Lesser General Public License as published by the Free # # Software Foundation, either version 3 of the License, or (at your option) # # any later version. # # # # PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY # # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # # details. # # # # You should have received a copy of the GNU Lesser General Public License # # along with PyGithub. If not, see . # # # ################################################################################ from typing import Any, Callable, Dict, Generic, Iterator, List, Optional, Type, TypeVar, Union from urllib.parse import parse_qs from github.GithubObject import GithubObject from github.Requester import Requester T = TypeVar("T", bound=GithubObject) class PaginatedListBase(Generic[T]): __elements: List[T] def _couldGrow(self) -> bool: raise NotImplementedError def _fetchNextPage(self) -> List[T]: raise NotImplementedError def __init__(self, elements: Optional[List[T]] = None) -> None: self.__elements = [] if elements is None else elements def __getitem__(self, index: Union[int, slice]) -> Any: assert isinstance(index, (int, slice)) if isinstance(index, int): self.__fetchToIndex(index) return self.__elements[index] else: return self._Slice(self, index) def __iter__(self) -> Iterator[T]: yield from self.__elements while self._couldGrow(): newElements = self._grow() yield from newElements def _isBiggerThan(self, index: int) -> bool: return len(self.__elements) > index or self._couldGrow() def __fetchToIndex(self, index: int) -> None: while len(self.__elements) <= index and self._couldGrow(): self._grow() def _grow(self) -> List[T]: newElements = self._fetchNextPage() self.__elements += newElements return newElements class _Slice: def __init__(self, theList: "PaginatedListBase[T]", theSlice: slice): self.__list = theList self.__start = theSlice.start or 0 self.__stop = theSlice.stop self.__step = theSlice.step or 1 def __iter__(self) -> Iterator[T]: index = self.__start while not self.__finished(index): if self.__list._isBiggerThan(index): yield self.__list[index] index += self.__step else: return def __finished(self, index: int) -> bool: return self.__stop is not None and index >= self.__stop class PaginatedList(PaginatedListBase[T]): """ This class abstracts the `pagination of the API `_. You can simply enumerate through instances of this class:: for repo in user.get_repos(): print(repo.name) If you want to know the total number of items in the list:: print(user.get_repos().totalCount) You can also index them or take slices:: second_repo = user.get_repos()[1] first_repos = user.get_repos()[:10] If you want to iterate in reversed order, just do:: for repo in user.get_repos().reversed: print(repo.name) And if you really need it, you can explicitly access a specific page:: some_repos = user.get_repos().get_page(0) some_other_repos = user.get_repos().get_page(3) """ def __init__( self, contentClass: Type[T], requester: Requester, firstUrl: str, firstParams: Any, headers: Optional[Dict[str, str]] = None, list_item: str = "items", total_count_item: str = "total_count", firstData: Optional[Any] = None, firstHeaders: Optional[Dict[str, Union[str, int]]] = None, attributesTransformer: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, ): self.__requester = requester self.__contentClass = contentClass self.__firstUrl = firstUrl self.__firstParams = firstParams or () self.__nextUrl = firstUrl self.__nextParams = firstParams or {} self.__headers = headers self.__list_item = list_item self.__total_count_item = total_count_item if self.__requester.per_page != 30: self.__nextParams["per_page"] = self.__requester.per_page self._reversed = False self.__totalCount: Optional[int] = None self._attributesTransformer = attributesTransformer first_page = [] if firstData is not None and firstHeaders is not None: first_page = self._getPage(firstData, firstHeaders) super().__init__(first_page) def _transformAttributes(self, element: Dict[str, Any]) -> Dict[str, Any]: if self._attributesTransformer is None: return element return self._attributesTransformer(element) @property def totalCount(self) -> int: if not self.__totalCount: params = {} if self.__nextParams is None else self.__nextParams.copy() # set per_page = 1 so the totalCount is just the number of pages params.update({"per_page": 1}) headers, data = self.__requester.requestJsonAndCheck( "GET", self.__firstUrl, parameters=params, headers=self.__headers ) if "link" not in headers: if data and "total_count" in data: self.__totalCount = data["total_count"] elif data: if isinstance(data, dict): data = data[self.__list_item] self.__totalCount = len(data) else: self.__totalCount = 0 else: links = self.__parseLinkHeader(headers) lastUrl = links.get("last") if lastUrl: self.__totalCount = int(parse_qs(lastUrl)["page"][0]) else: self.__totalCount = 0 return self.__totalCount # type: ignore def _getLastPageUrl(self) -> Optional[str]: headers, data = self.__requester.requestJsonAndCheck( "GET", self.__firstUrl, parameters=self.__nextParams, headers=self.__headers ) links = self.__parseLinkHeader(headers) return links.get("last") @property def reversed(self) -> "PaginatedList[T]": r = PaginatedList( self.__contentClass, self.__requester, self.__firstUrl, self.__firstParams, self.__headers, self.__list_item, attributesTransformer=self._attributesTransformer, ) r.__reverse() return r def __reverse(self) -> None: self._reversed = True lastUrl = self._getLastPageUrl() if lastUrl: self.__nextUrl = lastUrl def _couldGrow(self) -> bool: return self.__nextUrl is not None def _fetchNextPage(self) -> List[T]: headers, data = self.__requester.requestJsonAndCheck( "GET", self.__nextUrl, parameters=self.__nextParams, headers=self.__headers ) data = data if data else [] return self._getPage(data, headers) def _getPage(self, data: Any, headers: Dict[str, Any]) -> List[T]: self.__nextUrl = None # type: ignore if len(data) > 0: links = self.__parseLinkHeader(headers) if self._reversed: if "prev" in links: self.__nextUrl = links["prev"] elif "next" in links: self.__nextUrl = links["next"] self.__nextParams = None if self.__list_item in data: self.__totalCount = data.get(self.__total_count_item) data = data[self.__list_item] content = [ self.__contentClass(self.__requester, headers, self._transformAttributes(element), completed=False) for element in data if element is not None ] if self._reversed: return content[::-1] return content def __parseLinkHeader(self, headers: Dict[str, str]) -> Dict[str, str]: links = {} if "link" in headers: linkHeaders = headers["link"].split(", ") for linkHeader in linkHeaders: url, rel, *rest = linkHeader.split("; ") url = url[1:-1] rel = rel[5:-1] links[rel] = url return links def get_page(self, page: int) -> List[T]: params = dict(self.__firstParams) if page != 0: params["page"] = page + 1 if self.__requester.per_page != 30: params["per_page"] = self.__requester.per_page headers, data = self.__requester.requestJsonAndCheck( "GET", self.__firstUrl, parameters=params, headers=self.__headers ) if self.__list_item in data: self.__totalCount = data.get("total_count") data = data[self.__list_item] return [ self.__contentClass(self.__requester, headers, self._transformAttributes(element), completed=False) for element in data ] @classmethod def override_attributes(cls, overrides: Dict[str, Any]) -> Callable[[Dict[str, Any]], Dict[str, Any]]: def attributes_transformer(element: Dict[str, Any]) -> Dict[str, Any]: # Recursively merge overrides with attributes, overriding attributes with overrides element = cls.merge_dicts(element, overrides) return element return attributes_transformer @classmethod def merge_dicts(cls, d1: Dict[str, Any], d2: Dict[str, Any]) -> Dict[str, Any]: # clone d1 d1 = d1.copy() for k, v in d2.items(): if isinstance(v, dict): d1[k] = cls.merge_dicts(d1.get(k, {}), v) else: d1[k] = v return d1