Merge PaginatedList.pyi back to source (#2555)

Co-authored-by: Enrico Minack <github@enrico.minack.dev>
This commit is contained in:
Trim21
2023-06-20 08:13:20 +02:00
committed by GitHub
co-authored by Enrico Minack
parent b6258f4be9
commit cb50dec5ba
3 changed files with 42 additions and 72 deletions
+40 -28
View File
@@ -34,15 +34,28 @@
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################
from typing import Any, 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
class PaginatedListBase:
def __init__(self):
self.__elements = list()
self.__elements = []
def __getitem__(self, index):
def __getitem__(self, index: Union[int, slice]) -> Any:
assert isinstance(index, (int, slice))
if isinstance(index, int):
self.__fetchToIndex(index)
@@ -50,32 +63,32 @@ class PaginatedListBase:
else:
return self._Slice(self, index)
def __iter__(self):
def __iter__(self) -> Iterator[T]:
yield from self.__elements
while self._couldGrow():
newElements = self._grow()
yield from newElements
def _isBiggerThan(self, index):
def _isBiggerThan(self, index: int) -> bool:
return len(self.__elements) > index or self._couldGrow()
def __fetchToIndex(self, index):
while len(self.__elements) <= index and self._couldGrow():
self._grow()
def _grow(self):
def _grow(self) -> List[T]:
newElements = self._fetchNextPage()
self.__elements += newElements
return newElements
class _Slice:
def __init__(self, theList, theSlice):
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):
def __iter__(self) -> Iterator[T]:
index = self.__start
while not self.__finished(index):
if self.__list._isBiggerThan(index):
@@ -84,11 +97,11 @@ class PaginatedListBase:
else:
return
def __finished(self, index):
def __finished(self, index: int) -> bool:
return self.__stop is not None and index >= self.__stop
class PaginatedList(PaginatedListBase):
class PaginatedList(PaginatedListBase[T]):
"""
This class abstracts the `pagination of the API <https://docs.github.com/en/rest/guides/traversing-with-pagination>`_.
@@ -119,12 +132,12 @@ class PaginatedList(PaginatedListBase):
def __init__(
self,
contentClass,
requester,
firstUrl,
firstParams,
headers=None,
list_item="items",
contentClass: Type[T],
requester: Requester,
firstUrl: str,
firstParams: Any,
headers: Optional[Dict[str, str]] = None,
list_item: str = "items",
):
super().__init__()
self.__requester = requester
@@ -138,10 +151,10 @@ class PaginatedList(PaginatedListBase):
if self.__requester.per_page != 30:
self.__nextParams["per_page"] = self.__requester.per_page
self._reversed = False
self.__totalCount = None
self.__totalCount: Optional[int] = None
@property
def totalCount(self):
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
@@ -165,18 +178,17 @@ class PaginatedList(PaginatedListBase):
self.__totalCount = int(parse_qs(lastUrl)["page"][0])
else:
self.__totalCount = 0
return self.__totalCount
return self.__totalCount # type: ignore
def _getLastPageUrl(self):
def _getLastPageUrl(self) -> Optional[str]:
headers, data = self.__requester.requestJsonAndCheck(
"GET", self.__firstUrl, parameters=self.__nextParams, headers=self.__headers
)
links = self.__parseLinkHeader(headers)
lastUrl = links.get("last")
return lastUrl
return links.get("last")
@property
def reversed(self):
def reversed(self) -> "PaginatedList[T]":
r = PaginatedList(
self.__contentClass,
self.__requester,
@@ -197,13 +209,13 @@ class PaginatedList(PaginatedListBase):
def _couldGrow(self):
return self.__nextUrl is not None
def _fetchNextPage(self):
def _fetchNextPage(self) -> List[T]:
headers, data = self.__requester.requestJsonAndCheck(
"GET", self.__nextUrl, parameters=self.__nextParams, headers=self.__headers
)
data = data if data else []
self.__nextUrl = None
self.__nextUrl = None # type: ignore
if len(data) > 0:
links = self.__parseLinkHeader(headers)
if self._reversed:
@@ -226,7 +238,7 @@ class PaginatedList(PaginatedListBase):
return content[::-1]
return content
def __parseLinkHeader(self, headers):
def __parseLinkHeader(self, headers: Dict[str, str]) -> Dict[str, str]:
links = {}
if "link" in headers:
linkHeaders = headers["link"].split(", ")
@@ -237,7 +249,7 @@ class PaginatedList(PaginatedListBase):
links[rel] = url
return links
def get_page(self, page):
def get_page(self, page: int) -> List[T]:
params = dict(self.__firstParams)
if page != 0:
params["page"] = page + 1
-42
View File
@@ -1,42 +0,0 @@
from typing import Any, Dict, Generic, Iterator, List, Optional, Type, TypeVar, Union
from github.ContentFile import ContentFile
from github.GithubObject import GithubObject
from github.Issue import Issue
from github.NamedUser import NamedUser
from github.Requester import Requester
T = TypeVar("T", bound=GithubObject)
class PaginatedListBase(Generic[T]):
def __getitem__(self, index: Union[int, slice]) -> Any: ...
def __init__(self) -> None: ...
def __iter__(self) -> Iterator[T]: ...
def _grow(self) -> Any: ...
def _isBiggerThan(self, index: int) -> bool: ...
class _Slice:
def __init__(self, theList: PaginatedList[T], theSlice: slice) -> None: ...
def __iter__(self) -> Iterator[T]: ...
def __finished(self, index: int) -> bool: ...
class PaginatedList(PaginatedListBase[T]):
def __init__(
self,
contentClass: Type[T],
requester: Requester,
firstUrl: str,
firstParams: Any,
headers: Optional[Dict[str, str]] = ...,
list_item: str = ...,
) -> None: ...
def __reverse(self) -> None: ...
def __parseLinkHeader(self, headers: Dict[str, str]) -> None: ...
def _couldGrow(self) -> bool: ...
def _fetchNextPage(self) -> Any: ...
def _getLastPageUrl(self) -> Optional[str]: ...
def get_page(self, page: int) -> List[Any]: ...
@property
def reversed(self) -> PaginatedList[T]: ...
@property
def totalCount(self) -> int: ...
+2 -2
View File
@@ -426,7 +426,7 @@ class Requester:
parameters: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
input: Optional[Any] = None,
) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]:
) -> Tuple[Dict[str, Any], Any]:
return self.__check(
*self.requestJson(
verb, url, parameters, headers, input, self.__customConnection(url)
@@ -469,7 +469,7 @@ class Requester:
status: int,
responseHeaders: Dict[str, Any],
output: str,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
) -> Tuple[Dict[str, Any], Any]:
data = self.__structuredFromJson(output)
if status >= 400:
raise self.__createException(status, responseHeaders, data)