diff --git a/canvasapi/paginated_list.py b/canvasapi/paginated_list.py index 0183f93d..4a32cf96 100644 --- a/canvasapi/paginated_list.py +++ b/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 \ `_. + :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 "".format(self._content_class.__name__) + return f"" 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) diff --git a/tests/test_paginated_list.py b/tests/test_paginated_list.py index 9783e314..f3ecd790 100644 --- a/tests/test_paginated_list.py +++ b/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"))