Issue/539 typed paginated list

canvasapi/paginated_list.py

@@ -1,53 +1,75 @@
+from __future__ import annotations
+
 import re
+import typing as t
+from itertools import islice
+
+from requests.models import Response
+
+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>`_.
+    :param content_class: The expected type to return in the list.
+    :type content_class: class
+    :param requester: The requester to pass HTTP requests through.
+    :type requester: :class:`canvasapi.requester.Requester`
+    :param request_method: HTTP request method
+    :type request_method: str
+    :param first_url: Canvas endpoint for the initial request
+    :type first_url: str
+    :param extra_attribs: Extra data to include in the request
+    :type extra_attribs: dict
+    :param _root: Specify a nested property from Canvas to use for the resulting list.
+    :type _root: str
+    :param _url_override: "new_quizzes" or "graphql" for specific Canvas endpoints.
+    Other URLs may be specified for third-party requests.
+    :type _url_override: str
+    :rtype: :class:`canvasapi.paginated_list.PaginatedList` of type content_class
     """
 
-    def __getitem__(self, index):
-        assert isinstance(index, (int, slice))
+    @t.overload
+    def __getitem__(self, index: int) -> ContentClass:  # pragma: no cover
+        ...
+
+    @t.overload
+    def __getitem__(self, index: slice) -> t.List[ContentClass]:  # pragma: no cover
+        ...
+
+    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:
-                raise IndexError("Cannot negative index a PaginatedList")
-            self._get_up_to_index(index)
-            return self._elements[index]
-        else:
-            return self._Slice(self, index)
+                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]
 
     def __init__(
         self,
-        content_class,
-        requester,
-        request_method,
-        first_url,
-        extra_attribs=None,
-        _root=None,
-        _url_override=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,
+        _url_override: t.Optional[str] = None,
+        **kwargs: t.Any,
     ):
-        """
-        :param content_class: The expected type to return in the list.
-        :type content_class: class
-        :param requester: The requester to pass HTTP requests through.
-        :type requester: :class:`canvasapi.requester.Requester`
-        :param request_method: HTTP request method
-        :type request_method: str
-        :param first_url: Canvas endpoint for the initial request
-        :type first_url: str
-        :param extra_attribs: Extra data to include in the request
-        :type extra_attribs: dict
-        :param _root: Specify a nested property from Canvas to use for the resulting list.
-        :type _root: str
-        :param _url_override: "new_quizzes" or "graphql" for specific Canvas endpoints.
-        Other URLs may be specified for third-party requests.
-        :type _url_override: str
-        :rtype: :class:`canvasapi.paginated_list.PaginatedList` of type content_class
-        """
-        self._elements = list()
 
+        self._elements: list[ContentClass] = []
         self._requester = requester
         self._content_class = content_class
         self._first_url = first_url
@@ -60,103 +82,81 @@ def __init__(
         self._root = _root
         self._url_override = _url_override
 
-    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 __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)
 
     def __repr__(self):
-        return "<PaginatedList of type {}>".format(self._content_class.__name__)
+        return f"<PaginatedList of type {self._content_class.__name__}>"
 
     def _get_next_page(self):
-        response = self._requester.request(
+        data = self._request_next_page()
+        if self._root is not None:
+            try:
+                data = data[self._root]
+            except KeyError:
+                raise ValueError(
+                    "The specified _root value was not found in the response."
+                )
+        return [self._init_content_class(elem) for elem in data]
+
+    def _has_next_page(self):
+        """Check whether another page of results can be requested."""
+        return self._next_url is not None
+
+    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,
             _url=self._url_override,
             **self._next_params,
         )
-        data = response.json()
-        self._next_url = None
+        self._next_params = {}
         # Check the response headers first. This is the normal Canvas convention
         # for pagination, but there are endpoints which return a `meta` property
         # for pagination instead.
         # See https://github.com/ucfopen/canvasapi/discussions/605
         if response.links:
             next_link = response.links.get("next")
-        elif isinstance(data, dict) and "meta" in data:
+        elif isinstance(response, Response) and "meta" in response.json():
+            data = response.json()
             # requests parses headers into dicts, this returns the same
             # structure so the regex will still work.
             try:
-                next_link = {"url": data["meta"]["pagination"]["next"], "rel": "next"}
+                next_link = {
+                    "url": data["meta"]["pagination"]["next"],
+                    "rel": "next",
+                }
             except KeyError:
                 next_link = None
         else:
             next_link = None
 
-        regex = r"{}(.*)".format(re.escape(self._requester.base_url))
-
-        self._next_url = (
-            re.search(regex, next_link["url"]).group(1) if next_link else None
-        )
-
-        self._next_params = {}
+        self._set_next_url(next_link)
+        response_json: t.Any = response.json()
+        return response_json
 
-        content = []
-
-        if self._root:
-            try:
-                data = data[self._root]
-            except KeyError:
-                raise ValueError(
-                    "The key <{}> does not exist in the response.".format(self._root)
-                )
-
-        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):
-        return self._next_url is not None
+    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
 
-    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
+        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):
@@ -173,39 +177,6 @@ def test_root_element(self, m):
 
         self.assertIsInstance(pag_list[0], EnrollmentTerm)
 
-    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
-
-        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]
-
-    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
-
-        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]
-
-    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
-
-        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]
-
     def test_paginated_list_no_header(self, m):
         register_uris(
             {"paginated_list": ["no_header_4_2_pages_p1", "no_header_4_2_pages_p2"]}, m
@@ -233,3 +204,31 @@ def test_paginated_list_no_header_no_next(self, m):
         self.assertIsInstance(pag_list, PaginatedList)
         self.assertEqual(len(list(pag_list)), 2)
         self.assertIsInstance(pag_list[0], User)
+
+    # Negative indicies
+    def test_negative_index(self, m):
+        # Regression test for https://github.com/ucfopen/canvasapi/issues/305
+        # 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]
+        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 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]
+        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 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]
+
+        self.assertEqual([elem.id for elem in pag_list[:-1]], list("123"))