From 24f672b836e1c3bb673eaa6702c758b866ad1594 Mon Sep 17 00:00:00 2001 From: Garrett Shipley Date: Mon, 25 Jul 2022 10:05:35 -0400 Subject: [PATCH 1/3] 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/3] 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/3] 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]