Merge pull request #17 from RedHatProductSecurity/smart-pagination

Implement better pagination

CHANGELOG.md

@@ -11,6 +11,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
+### Changed
+- implement smarter pagination
 
 ## [1.3.4] - 2023-05-04
 

TUTORIAL.md

@@ -96,6 +96,20 @@ See `/GET /api/{api_version}/components` in [API docs](openapi_schema.yml) for m
   select_components = session.components.retrieve_list(arch="x86_64")
 ```
 
+#### components.retrieve_list_iterator
+
+Retrieve a list of Components. Handles the pagination and returns the generator of individual resource entities.
+
+See `/GET /api/{api_version}/components` in [API docs](openapi_schema.yml) for more details (query parameters, response format, etc.)
+```python
+  all_components = session.components.retrieve_list_iterator()
+  for component in all_components:
+    do_calc(component)
+
+  for component in session.components.retrieve_list_iterator(arch="x86_64"):
+    print(component.arch)
+```
+
 #### components.retrieve
 
 Retrieve a single Component with specified `id`.

component_registry_bindings/constants.py

@@ -1,7 +1,14 @@
+"""
+component-registry-bindings constants
+"""
+
 from typing import Dict, List
 
 COMPONENT_REGISTRY_API_VERSION: str = "v1"
-COMPONENT_REGISTRY_BINDINGS_USERAGENT: str = "component-registry-bindings-1.3.4"
+COMPONENT_REGISTRY_BINDINGS_VERSION: str = "1.3.4"
+COMPONENT_REGISTRY_BINDINGS_USERAGENT: str = (
+    f"component-registry-bindings-{COMPONENT_REGISTRY_BINDINGS_VERSION}"
+)
 COMPONENT_REGISTRY_BINDINGS_API_PATH: str = (
     f".bindings.python_client.api.{COMPONENT_REGISTRY_API_VERSION}"
 )

component_registry_bindings/exceptions.py

@@ -0,0 +1,11 @@
+"""
+component-registry-bindings exceptions
+"""
+
+
+class ComponentRegistryBindingsException(Exception):
+    """Base component-registry-bindings exception"""
+
+
+class OperationUnsupported(ComponentRegistryBindingsException):
+    """Session operation is unsupported exception"""

component_registry_bindings/iterators.py

@@ -0,0 +1,110 @@
+"""
+component-registry iterators
+"""
+
+import re
+from functools import partial
+from typing import Callable, Optional
+
+from .exceptions import ComponentRegistryBindingsException
+
+
+class Paginator:
+    """
+    Iterator for handling API pagination.
+
+    Receives either starting limit and offset together with the retreive list function
+    or already existing response from which it should continue.
+
+    It keeps calling `.next()` response until pages are exhausted.
+    """
+
+    def __init__(
+        self,
+        *args,
+        retrieve_list_fn: Optional[Callable] = None,
+        limit: int = 100,
+        offset: int = 0,
+        init_response=None,
+        **kwargs,
+    ):
+
+        if not init_response and not retrieve_list_fn:
+            raise ComponentRegistryBindingsException(
+                (
+                    "Paginator needs either initial response or session"
+                    "operation function to obtain the initial response."
+                )
+            )
+
+        self.retrieve_list_fn = retrieve_list_fn
+
+        # initial starting data
+        self.__init_limit = limit
+        self.__init_offset = offset
+        self.__init_response = init_response
+
+        # current response page
+        self.current_response = init_response
+
+        # request arguments
+        self.args = args
+        self.kwargs = kwargs
+
+    def __iter__(self):
+        # restore initial response
+        self.current_response = self.__init_response
+        return self
+
+    def __next__(self):
+        if self.current_response is None:
+
+            # no current response page - obtain the first page
+            response = self.retrieve_list_fn(
+                *self.args,
+                limit=self.__init_limit,
+                offset=self.__init_offset,
+                **self.kwargs,
+            )
+            response = self.make_response_iterable(
+                response, self.retrieve_list_fn, *self.args, **self.kwargs
+            )
+            self.current_response = response
+            return response
+        else:
+
+            # existing current response - call next page
+            response = self.current_response.next()
+            if response is not None:
+                self.current_response = response
+                return response
+            else:
+                raise StopIteration
+
+    @staticmethod
+    def make_response_iterable(response, retrieve_list_fn, *args, **kwargs):
+        """
+        Populate next, prev and iterator helper methods for paginated responses
+        """
+
+        response.iterator = Paginator(
+            init_response=response,
+        )
+
+        for param_name, func_name in (("next_", "next"), ("previous", "prev")):
+            kwargs.pop("limit", None)
+            kwargs.pop("offset", None)
+            param = getattr(response, param_name, None)
+            if param is None:
+                setattr(response, func_name, lambda: None)
+            else:
+                limit = re.search("limit=(\d+)", param)
+                if limit is not None:
+                    kwargs["limit"] = limit.group(1)
+                offset = re.search("offset=(\d+)", param)
+                if offset is not None:
+                    kwargs["offset"] = offset.group(1)
+
+                setattr(response, func_name, partial(retrieve_list_fn, *args, **kwargs))
+
+        return response