Merge PaginatedList.pyi back to source

github/PaginatedList.py

@@ -34,48 +34,61 @@
 # 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)
             return self.__elements[index]
         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 @@ def __iter__(self):
                 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 @@ def __init__(
         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 @@ def totalCount(self):
                     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 @@ def __reverse(self):
     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 @@ def _fetchNextPage(self):
             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 @@ def __parseLinkHeader(self, headers):
                 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

github/Requester.py

@@ -426,7 +426,7 @@ def requestJsonAndCheck(
         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 @@ def __check(
         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)