Type hinting paginated lists

canvasapi/paginated_list.py

@@ -1,34 +1,34 @@
+from __future__ import annotations
+
 import re
+import typing as t
+from itertools import islice
+
+if t.TYPE_CHECKING:
+    from canvasapi.requester import Requester  # pragma: no cover
+
+
+ContentClass = t.TypeVar("ContentClass")
 
 
-class PaginatedList(object):
+class PaginatedList(t.Generic[ContentClass]):
     """
     Abstracts `pagination of Canvas API \
     <https://canvas.instructure.com/doc/api/file.pagination.html>`_.
     """
 
-    def __getitem__(self, index):
-        assert isinstance(index, (int, slice))
-        if isinstance(index, int):
-            if index < 0:
-                raise IndexError("Cannot negative index a PaginatedList")
-            self._get_up_to_index(index)
-            return self._elements[index]
-        else:
-            return self._Slice(self, index)
-
     def __init__(
         self,
-        content_class,
-        requester,
-        request_method,
-        first_url,
-        extra_attribs=None,
-        _root=None,
-        **kwargs
+        content_class: type[ContentClass],
+        requester: "Requester",
+        request_method: str,
+        first_url: str,
+        extra_attribs: t.Optional[t.Dict[str, t.Any]] = None,
+        _root: t.Optional[str] = None,
+        **kwargs: t.Any,
     ):
 
-        self._elements = list()
+        self._elements: list[ContentClass] = []
 
         self._requester = requester
         self._content_class = content_class
@@ -41,85 +41,82 @@ def __init__(
         self._request_method = request_method
         self._root = _root
 
-    def __iter__(self):
-        for element in self._elements:
-            yield element
-        while self._has_next():
-            new_elements = self._grow()
-            for element in new_elements:
-                yield element
-
-    def __repr__(self):
-        return "<PaginatedList of type {}>".format(self._content_class.__name__)
-
-    def _get_next_page(self):
-        response = self._requester.request(
-            self._request_method, self._next_url, **self._next_params
-        )
-        data = response.json()
-        self._next_url = None
+    def __iter__(self) -> t.Generator[ContentClass, None, None]:
+        if self._has_next_page():
+            current_index = 0
+            while self._has_next_page():
+                self._elements.extend(self._get_next_page())
+                for index, element in enumerate(self._elements):
+                    if index == current_index:
+                        yield element
+                        current_index += 1
+        else:
+            yield from iter(self._elements)
 
-        next_link = response.links.get("next")
-        regex = r"{}(.*)".format(re.escape(self._requester.base_url))
+    @t.overload
+    def __getitem__(self, index: int) -> ContentClass:  # pragma: no cover
+        ...
 
-        self._next_url = (
-            re.search(regex, next_link["url"]).group(1) if next_link else None
-        )
+    @t.overload
+    def __getitem__(self, index: slice) -> t.List[ContentClass]:  # pragma: no cover
+        ...
 
-        self._next_params = {}
+    def __getitem__(self, index: int | slice):
+        assert isinstance(
+            index, (int, slice)
+        ), "`index` must be either an integer or a slice."
+        if isinstance(index, int):
+            if index < 0:
+                return list(self)[index]
+            return list(islice(self, index + 1))[index]
+        # if no negatives, islice can be used
+        if not any(
+            v is not None and v < 0 for v in (index.start, index.stop, index.step)
+        ):
+            return list(islice(self, index.start, index.stop, index.step))
+        return list(self)[index]
 
-        content = []
+    def __repr__(self):
+        return f"<PaginatedList of type {self._content_class.__name__}>"
 
-        if self._root:
+    def _get_next_page(self):
+        data = self._request_next_page()
+        if self._root is not None:
             try:
                 data = data[self._root]
             except KeyError:
                 # TODO: Fix this message to make more sense to an end user.
                 raise ValueError("Invalid root value specified.")
+        return [self._init_content_class(elem) for elem in data]
 
-        for element in data:
-            if element is not None:
-                element.update(self._extra_attribs)
-                content.append(self._content_class(self._requester, element))
-
-        return content
-
-    def _get_up_to_index(self, index):
-        while len(self._elements) <= index and self._has_next():
-            self._grow()
-
-    def _grow(self):
-        new_elements = self._get_next_page()
-        self._elements += new_elements
-        return new_elements
-
-    def _has_next(self):
+    def _has_next_page(self):
+        """Check whether another page of results can be requested."""
         return self._next_url is not None
 
-    def _is_larger_than(self, index):
-        return len(self._elements) > index or self._has_next()
-
-    class _Slice(object):
-        def __init__(self, the_list, the_slice):
-            self._list = the_list
-            self._start = the_slice.start or 0
-            self._stop = the_slice.stop
-            self._step = the_slice.step or 1
-
-            if self._start < 0 or self._stop < 0:
-                raise IndexError("Cannot negative index a PaginatedList slice")
-
-        def __iter__(self):
-            index = self._start
-            while not self._finished(index):
-                if self._list._is_larger_than(index):
-                    try:
-                        yield self._list[index]
-                    except IndexError:
-                        return
-                    index += self._step
-                else:
-                    return
-
-        def _finished(self, index):
-            return self._stop is not None and index >= self._stop
+    def _init_content_class(self, attributes: t.Dict[str, t.Any]):
+        """Instantiate a new content class."""
+        return self._content_class(
+            self._requester, {**attributes, **self._extra_attribs}
+        )
+
+    def _request_next_page(self):
+        response: t.Any = self._requester.request(
+            self._request_method,
+            self._next_url,
+            **self._next_params,
+        )
+        self._next_params = {}
+        self._set_next_url(response.links.get("next"))
+        response_json: t.Any = response.json()
+        return response_json
+
+    def _set_next_url(self, next_link: t.Optional[t.Dict[str, str]]):
+        """Set the next url to request, if on exists."""
+        if next_link is None:
+            self._next_url = None
+            return
+
+        regex = rf"{re.escape(self._requester.base_url)}(.*)"
+        match = re.search(regex, next_link["url"])
+        if match is not None:
+            self._next_url = match.group(1)

tests/test_paginated_list.py

@@ -151,7 +151,11 @@ def test_root_element_incorrect(self, m):
         register_uris({"account": ["get_enrollment_terms"]}, m)
 
         pag_list = PaginatedList(
-            EnrollmentTerm, self.requester, "GET", "accounts/1/terms", _root="wrong"
+            EnrollmentTerm,
+            self.requester,
+            "GET",
+            "accounts/1/terms",
+            _root="wrong",
         )
 
         with self.assertRaises(ValueError):
@@ -172,33 +176,27 @@ def test_root_element(self, m):
 
     def test_negative_index(self, m):
         # Regression test for https://github.com/ucfopen/canvasapi/issues/305
-        # Ensure that we can't use negative indexing, even after loading a page
+        # Ensure that we can use negative indexing, even after loading a page
 
         register_uris({"paginated_list": ["4_2_pages_p1", "4_2_pages_p2"]}, m)
         pag_list = PaginatedList(User, self.requester, "GET", "four_objects_two_pages")
         pag_list[0]
-
-        with self.assertRaises(IndexError):
-            pag_list[-1]
+        self.assertEqual(pag_list[-1].id, "4")
 
     def test_negative_index_for_slice_start(self, m):
         # Regression test for https://github.com/ucfopen/canvasapi/issues/305
-        # Ensure that we can't slice using a negative index as the start item
+        # Ensure that we can slice using a negative index as the start item
 
         register_uris({"paginated_list": ["4_2_pages_p1", "4_2_pages_p2"]}, m)
         pag_list = PaginatedList(User, self.requester, "GET", "four_objects_two_pages")
         pag_list[0]
-
-        with self.assertRaises(IndexError):
-            pag_list[-1:1]
+        self.assertEqual(pag_list[-1:1], [])
 
     def test_negative_index_for_slice_end(self, m):
         # Regression test for https://github.com/ucfopen/canvasapi/issues/305
-        # Ensure that we can't slice using a negative index as the end item
+        # Ensure that we can slice using a negative index as the end item
 
         register_uris({"paginated_list": ["4_2_pages_p1", "4_2_pages_p2"]}, m)
         pag_list = PaginatedList(User, self.requester, "GET", "four_objects_two_pages")
         pag_list[0]
-
-        with self.assertRaises(IndexError):
-            pag_list[:-1]
+        self.assertEqual([elem.id for elem in pag_list[:-1]], list("123"))