chore: remove 'while True' because why not

Author: aaxelb

tests/share/search/index_strategy/_common_trovesearch_tests.py

@@ -117,10 +117,10 @@ def test_cardsearch_pagination(self):
             }))
         self._index_indexcards(_cards)
         # gather all pages results:
-        _querystring: str = f'page[size]={_page_size}'
+        _querystring: str | None = f'page[size]={_page_size}'
         _result_iris: set[str] = set()
         _page_count = 0
-        while True:
+        while _querystring is not None:
             _cardsearch_handle = self.index_strategy.pls_handle_cardsearch(
                 CardsearchParams.from_querystring(_querystring),
             )
@@ -133,9 +133,11 @@ def test_cardsearch_pagination(self):
             _result_iris.update(_page_iris)
             _page_count += 1
             _next_cursor = _cardsearch_handle.cursor.next_cursor()
-            if _next_cursor is None:
-                break
-            _querystring = urlencode({'page[cursor]': _next_cursor.as_queryparam_value()})
+            _querystring = (
+                urlencode({'page[cursor]': _next_cursor.as_queryparam_value()})
+                if _next_cursor is not None
+                else None  # done
+            )
         self.assertEqual(_page_count, math.ceil(_total_count / _page_size))
         self.assertEqual(_result_iris, _expected_iris)
 

tests/share/test_oaipmh_trove.py

@@ -232,11 +232,9 @@ def _assert_full_list(self, verb, params, request_method, expected_count, page_s
         pages = 0
         count = 0
         token = None
-        while True:
-            if token:
-                parsed = oai_request({'verb': verb, 'resumptionToken': token}, request_method)
-            else:
-                parsed = oai_request({'verb': verb, 'metadataPrefix': 'oai_dc', **params}, request_method)
+        next_params: dict[str, str] | None = {'verb': verb, 'metadataPrefix': 'oai_dc', **params}
+        while next_params is not None:
+            parsed = oai_request(next_params, request_method)
             page = parsed.xpath('//oai:header/oai:identifier', namespaces=NAMESPACES)
             pages += 1
             count += len(page)
@@ -245,9 +243,10 @@ def _assert_full_list(self, verb, params, request_method, expected_count, page_s
             token = token[0].text
             if token:
                 assert len(page) == page_size
+                next_params = {'verb': verb, 'resumptionToken': token}
             else:
                 assert len(page) <= page_size
-                break
+                next_params = None  # done
 
         assert count == expected_count
         assert pages == math.ceil(expected_count / page_size)

trove/render/jsonapi.py

@@ -38,15 +38,11 @@
 
 def _resource_ids_defaultdict() -> defaultdict[Any, str]:
     _prefix = str(time.time_ns())
-    _ints = itertools.count()
-
-    def _iter_ids() -> Iterator[str]:
-        while True:
-            _id = next(_ints)
-            yield f'{_prefix}-{_id}'
-
-    _ids = _iter_ids()
-    return defaultdict(lambda: next(_ids))
+    _infinite_ids = (
+        f'{_prefix}-{_id}'
+        for _id in itertools.count()
+    )
+    return defaultdict(_infinite_ids.__next__)
 
 
 @dataclasses.dataclass

trove/util/django.py

@@ -16,18 +16,16 @@ def pk_chunked(queryset: QuerySet, chunksize: int) -> Generator[list]:
     '''
     _ordered_qs = queryset.order_by('pk')
     _prior_end_pk = None
-    while True:  # for each chunk:
-        _qs = (
-            _ordered_qs
-            if _prior_end_pk is None
-            else _ordered_qs.filter(pk__gt=_prior_end_pk)
-        )
+    _chunk_qs: QuerySet | None = _ordered_qs
+    while _chunk_qs is not None:  # for each chunk:
         # load primary key values only
-        _pks = list(_qs.values_list('pk', flat=True)[:chunksize])
-        if not _pks:
-            break  # done
-        _end_pk = _pks[-1]
-        if (_prior_end_pk is not None) and (_end_pk <= _prior_end_pk):
-            raise RuntimeError(f'sentinel pks not ascending?? got {_end_pk} after {_prior_end_pk}')
-        _prior_end_pk = _end_pk
-        yield _pks
+        _pks = list(_chunk_qs.values_list('pk', flat=True)[:chunksize])
+        if _pks:
+            _end_pk = _pks[-1]
+            if (_prior_end_pk is not None) and (_end_pk <= _prior_end_pk):
+                raise RuntimeError(f'sentinel pks not ascending?? got {_end_pk} after {_prior_end_pk}')
+            yield _pks
+            _prior_end_pk = _end_pk
+            _chunk_qs = _ordered_qs.filter(pk__gt=_prior_end_pk)
+        else:
+            _chunk_qs = None  # done