For APIs that may be paged, entries, keys, and values, make the paging opt-in given the performance considerations.

Author: rlubke

src/coherence/client.py

@@ -400,7 +400,7 @@ async def aggregate(
         """
 
     @abc.abstractmethod
-    def values(self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None) -> AsyncIterator[V]:
+    def values(self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None, by_page: bool = False) -> AsyncIterator[V]:
         """
         Return a Set of the values contained in this map that satisfy the criteria expressed by the filter.
         If no filter or comparator is specified, it returns a Set view of the values contained in this map.The
@@ -411,22 +411,26 @@ def values(self, filter: Optional[Filter] = None, comparator: Optional[Comparato
         :param filter: the Filter object representing the criteria that the entries of this map should satisfy
         :param comparator:  the Comparator object which imposes an ordering on entries in the resulting set; or null
          if the entries' natural ordering should be used
+        :param by_page: returns the keys in pages (transparently to the caller).  This option is only valid
+         if no filter or comparator is provided.
         :return: an AsyncIterator of MapEntry instances resolving to the values that satisfy the specified criteria
         """
 
     @abc.abstractmethod
-    def keys(self, filter: Optional[Filter] = None) -> AsyncIterator[K]:
+    def keys(self, filter: Optional[Filter] = None, by_page: bool = False) -> AsyncIterator[K]:
         """
         Return a set view of the keys contained in this map for entries that satisfy the criteria expressed by the
         filter.
 
         :param filter: the Filter object representing the criteria that the entries of this map should satisfy
+        :param by_page: returns the keys in pages (transparently to the caller).  This option is only valid
+         if no filter is provided.
         :return: an AsyncIterator of keys for entries that satisfy the specified criteria
         """
 
     @abc.abstractmethod
     def entries(
-        self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None
+        self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None, by_page: bool = False
     ) -> AsyncIterator[MapEntry[K, V]]:
         """
         Return a set view of the entries contained in this map that satisfy the criteria expressed by the filter.
@@ -435,6 +439,8 @@ def entries(
         :param filter: the Filter object representing the criteria that the entries of this map should satisfy
         :param comparator: the Comparator object which imposes an ordering on entries in the resulting set; or `None`
          if the entries' values natural ordering should be used
+        :param by_page: returns the keys in pages (transparently to the caller).  This option is only valid
+         if no filter or comparator is provided.
         :return: an AsyncIterator of MapEntry instances that satisfy the specified criteria
         """
 
@@ -664,10 +670,8 @@ async def aggregate(
         return cast(R, value)
 
     @_pre_call_cache
-    def values(self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None) -> AsyncIterator[V]:
-        # if there is no filter or no co, or the filter is an AlwaysFilter,
-        # obtain results by-page
-        if (comparator is None and filter is None) or (comparator is None and isinstance(filter, AlwaysFilter)):
+    def values(self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None, by_page: bool = False) -> AsyncIterator[V]:
+        if by_page and comparator is None and filter is None:
             return _PagedStream(self, _scalar_deserializer)
         else:
             r = self._request_factory.values_request(filter)
@@ -676,10 +680,8 @@ def values(self, filter: Optional[Filter] = None, comparator: Optional[Comparato
             return _Stream(self._request_factory.get_serializer(), stream, _scalar_producer)
 
     @_pre_call_cache
-    def keys(self, filter: Optional[Filter] = None) -> AsyncIterator[K]:
-        # if there is no filter, or the filter is an AlwaysFilter,
-        # obtain results by-page
-        if filter is None or isinstance(filter, AlwaysFilter):
+    def keys(self, filter: Optional[Filter] = None, by_page: bool = False) -> AsyncIterator[K]:
+        if by_page and filter is None:
             return _PagedStream(self, _scalar_deserializer, True)
         else:
             r = self._request_factory.keys_request(filter)
@@ -689,11 +691,9 @@ def keys(self, filter: Optional[Filter] = None) -> AsyncIterator[K]:
 
     @_pre_call_cache
     def entries(
-        self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None
+        self, filter: Optional[Filter] = None, comparator: Optional[Comparator] = None, by_page: bool = False
     ) -> AsyncIterator[MapEntry[K, V]]:
-        # if there is no filter and no comparator, or the filter is an AlwaysFilter,
-        # obtain results by-page
-        if (comparator is None and filter is None) or (comparator is None and isinstance(filter, AlwaysFilter)):
+        if by_page and comparator is None and filter is None:
             return _PagedStream(self, _entry_deserializer)
         else:
             r = self._request_factory.entries_request(filter, comparator)

tests/test_client.py

@@ -225,7 +225,7 @@ async def test_keys_paged(setup_and_teardown: NamedCache[str, str]) -> None:
 
     # Stream the keys and locally cache the results
     local_set: set[str] = set()
-    async for e in cache.keys():
+    async for e in cache.keys(by_page=True):
         local_set.add(e)
 
     assert len(local_set) == num_entries
@@ -268,7 +268,7 @@ async def test_entries_paged(setup_and_teardown: NamedCache[str, str]) -> None:
 
     # Stream the keys and locally cache the results
     local_dict: dict[str, str] = {}
-    async for e in cache.entries():
+    async for e in cache.entries(by_page=True):
         local_dict[e.key] = e.value
 
     assert len(local_dict) == num_entries
@@ -308,7 +308,7 @@ async def test_values_paged(setup_and_teardown: NamedCache[str, str]) -> None:
 
     # Stream the keys and locally cache the results
     local_list: list[str] = []
-    async for e in cache.values():
+    async for e in cache.values(by_page=True):
         local_list.append(e)
 
     assert len(local_list) == num_entries