From 24f672b836e1c3bb673eaa6702c758b866ad1594 Mon Sep 17 00:00:00 2001 From: Garrett Shipley Date: Mon, 25 Jul 2022 10:05:35 -0400 Subject: [PATCH 1/4] Added type hinting and negative indexing to PaginatedList --- canvasapi/paginated_list.py | 163 +++++++++++++++++------------------- 1 file changed, 75 insertions(+), 88 deletions(-) diff --git a/canvasapi/paginated_list.py b/canvasapi/paginated_list.py index 9be9fc3e..8b49629c 100644 --- a/canvasapi/paginated_list.py +++ b/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 \ `_. """ - 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,72 @@ 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 "".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"" - 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 = self._requester.request(self._request_method, self._next_url, **self._next_params) # type: ignore + 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) From 27c2fae0581f41da19d377c533b0322026a57ac3 Mon Sep 17 00:00:00 2001 From: Garrett Shipley Date: Mon, 25 Jul 2022 10:07:30 -0400 Subject: [PATCH 2/4] Updates test_paginated_list.py --- tests/test_paginated_list.py | 22 +++++++--------------- 1 file changed, 7 insertions(+), 15 deletions(-) diff --git a/tests/test_paginated_list.py b/tests/test_paginated_list.py index 3b9e604d..3b80fc2f 100644 --- a/tests/test_paginated_list.py +++ b/tests/test_paginated_list.py @@ -150,9 +150,7 @@ def test_repr(self, m): 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" - ) + pag_list = PaginatedList(EnrollmentTerm, self.requester, "GET", "accounts/1/terms", _root="wrong") with self.assertRaises(ValueError): pag_list[0] @@ -172,33 +170,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")) From 500f3950c3300f2faae10b0250d62c819eb668f5 Mon Sep 17 00:00:00 2001 From: Garrett Shipley Date: Mon, 25 Jul 2022 10:14:45 -0400 Subject: [PATCH 3/4] formatting --- canvasapi/paginated_list.py | 18 ++++++++++++++---- tests/test_paginated_list.py | 8 +++++++- 2 files changed, 21 insertions(+), 5 deletions(-) diff --git a/canvasapi/paginated_list.py b/canvasapi/paginated_list.py index 8b49629c..13a090c8 100644 --- a/canvasapi/paginated_list.py +++ b/canvasapi/paginated_list.py @@ -62,13 +62,17 @@ 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." + 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)): + 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] @@ -91,10 +95,16 @@ def _has_next_page(self): 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}) + return self._content_class( + self._requester, {**attributes, **self._extra_attribs} + ) def _request_next_page(self): - response = self._requester.request(self._request_method, self._next_url, **self._next_params) # type: ignore + 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() diff --git a/tests/test_paginated_list.py b/tests/test_paginated_list.py index 3b80fc2f..d75ab028 100644 --- a/tests/test_paginated_list.py +++ b/tests/test_paginated_list.py @@ -150,7 +150,13 @@ def test_repr(self, m): 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") + pag_list = PaginatedList( + EnrollmentTerm, + self.requester, + "GET", + "accounts/1/terms", + _root="wrong", + ) with self.assertRaises(ValueError): pag_list[0] From f8c79632352466ef62820af42e534d83f0af52dd Mon Sep 17 00:00:00 2001 From: bennettscience Date: Wed, 15 May 2024 14:46:54 -0400 Subject: [PATCH 4/4] Bring up to date with upstream/develop This brings the PR up to speed with the current `develop` branch on upstream. All tests passing, accounting for updates to `PaginatedList` and `Requester` since the original PR was introduced. --- canvasapi/paginated_list.py | 59 ++++++++++++++++++++---------------- tests/test_paginated_list.py | 56 +++++++++++++++++----------------- 2 files changed, 60 insertions(+), 55 deletions(-) diff --git a/canvasapi/paginated_list.py b/canvasapi/paginated_list.py index 3b5668ef..4a32cf96 100644 --- a/canvasapi/paginated_list.py +++ b/canvasapi/paginated_list.py @@ -4,6 +4,8 @@ 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 @@ -32,6 +34,29 @@ class PaginatedList(t.Generic[ContentClass]): :rtype: :class:`canvasapi.paginated_list.PaginatedList` of type content_class """ + @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: + 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: type[ContentClass], @@ -40,6 +65,7 @@ def __init__( 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, ): @@ -68,29 +94,6 @@ def __iter__(self) -> t.Generator[ContentClass, None, None]: else: yield from iter(self._elements) - @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: - 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 __repr__(self): return f"" @@ -100,7 +103,6 @@ def _get_next_page(self): try: data = data[self._root] except KeyError: - # TODO: Raise a better error message to the user. raise ValueError( "The specified _root value was not found in the response." ) @@ -120,6 +122,7 @@ def _request_next_page(self): response: t.Any = self._requester.request( self._request_method, self._next_url, + _url=self._url_override, **self._next_params, ) self._next_params = {} @@ -129,11 +132,15 @@ def _request_next_page(self): # 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: diff --git a/tests/test_paginated_list.py b/tests/test_paginated_list.py index 8ff602af..f3ecd790 100644 --- a/tests/test_paginated_list.py +++ b/tests/test_paginated_list.py @@ -177,35 +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 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] - - 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,4 +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"))