feat(pagination): Refactor FHIRSearch with iterable Bundle, add iter …

…to Bundle, move pagination logic to _utils.py, add tests

fhirclient/_utils.py

@@ -0,0 +1,133 @@
+import urllib
+from typing import Optional, Iterable
+
+import requests
+
+from typing import TYPE_CHECKING, Iterable
+
+if TYPE_CHECKING:
+    from fhirclient.models.bundle import Bundle
+
+
+# Use forward references to avoid circular imports
+def _fetch_next_page(bundle: 'Bundle') -> Optional['Bundle']:
+    """
+    Fetch the next page of results using the `next` link provided in the bundle.
+
+    Args:
+        bundle (Bundle): The FHIR Bundle containing the `next` link.
+
+    Returns:
+        Optional[Bundle]: The next page of results as a FHIR Bundle, or None if no "next" link is found.
+    """
+    next_link = _get_next_link(bundle)
+    if next_link:
+        sanitized_next_link = _sanitize_next_link(next_link)
+        return _execute_pagination_request(sanitized_next_link)
+    return None
+
+
+def _get_next_link(bundle: 'Bundle') -> Optional[str]:
+    """
+    Extract the `next` link from the Bundle's links.
+
+    Args:
+        bundle (Bundle): The FHIR Bundle containing pagination links.
+
+    Returns:
+        Optional[str]: The URL of the next page if available, None otherwise.
+    """
+    if not bundle.link:
+        return None
+
+    for link in bundle.link:
+        if link.relation == "next":
+            return link.url
+    return None
+
+
+def _sanitize_next_link(next_link: str) -> str:
+    """
+    Sanitize the `next` link to ensure it is safe to use.
+
+    Args:
+        next_link (str): The raw `next` link URL.
+
+    Returns:
+        str: The sanitized URL.
+
+    Raises:
+        ValueError: If the URL scheme or domain is invalid.
+    """
+    parsed_url = urllib.parse.urlparse(next_link)
+
+    # Validate scheme and netloc (domain)
+    if parsed_url.scheme not in ["http", "https"]:
+        raise ValueError("Invalid URL scheme in `next` link.")
+    if not parsed_url.netloc:
+        raise ValueError("Invalid URL domain in `next` link.")
+
+    # Additional sanitization if necessary, e.g., removing dangerous query parameters
+    query_params = urllib.parse.parse_qs(parsed_url.query)
+    sanitized_query = {k: v for k, v in query_params.items()}
+
+    # Rebuild the sanitized URL
+    sanitized_url = urllib.parse.urlunparse(
+        (
+            parsed_url.scheme,
+            parsed_url.netloc,
+            parsed_url.path,
+            parsed_url.params,
+            urllib.parse.urlencode(sanitized_query, doseq=True),
+            parsed_url.fragment,
+        )
+    )
+
+    return sanitized_url
+
+
+def _execute_pagination_request(sanitized_url: str) -> Optional['Bundle']:
+    """
+    Execute the request to retrieve the next page using the sanitized URL.
+
+    Args:
+        sanitized_url (str): The sanitized URL to fetch the next page.
+
+    Returns:
+        Optional[Bundle]: The next page of results as a FHIR Bundle, or None.
+
+    Raises:
+        HTTPError: If the request fails due to network issues or server errors.
+    """
+    from fhirclient.models.bundle import Bundle  # Import here to avoid circular import
+
+    try:
+        # Use requests.get directly to make the HTTP request
+        response = requests.get(sanitized_url)
+        response.raise_for_status()
+        next_bundle_data = response.json()
+        next_bundle = Bundle(next_bundle_data)
+
+        return next_bundle
+
+    except requests.exceptions.HTTPError as e:
+        # Handle specific HTTP errors as needed, possibly including retry logic
+        raise e
+
+
+def iter_pages(first_bundle: 'Bundle') -> Iterable['Bundle']:
+    """
+    Iterator that yields each page of results as a FHIR Bundle.
+
+    Args:
+        first_bundle (Optional[Bundle]): The first Bundle to start pagination.
+
+    Yields:
+        Bundle: Each page of results as a FHIR Bundle.
+    """
+    # Since _fetch_next_page can return None
+    bundle: Optional[Bundle] = first_bundle
+    while bundle:
+        yield bundle
+        bundle = _fetch_next_page(bundle)
+

fhirclient/client.py

@@ -1,10 +1,4 @@
 import logging
-import urllib
-from typing import Optional, Iterable
-
-import requests
-from .models.bundle import Bundle
-
 from .server import FHIRServer, FHIRUnauthorizedException, FHIRNotFoundException
 
 __version__ = '4.3.0'
@@ -244,119 +238,3 @@ def from_state(self, state):
 
     def save_state (self):
         self._save_func(self.state)
-
-    # MARK: Pagination
-    def _fetch_next_page(self, bundle: Bundle) -> Optional[Bundle]:
-        """
-        Fetch the next page of results using the `next` link provided in the bundle.
-
-        Args:
-            bundle (Bundle): The FHIR Bundle containing the `next` link.
-
-        Returns:
-            Optional[Bundle]: The next page of results as a FHIR Bundle, or None if no "next" link is found.
-        """
-        next_link = self._get_next_link(bundle)
-        if next_link:
-            sanitized_next_link = self._sanitize_next_link(next_link)
-            return self._execute_pagination_request(sanitized_next_link)
-        return None
-
-    def _get_next_link(self, bundle: Bundle) -> Optional[str]:
-        """
-        Extract the `next` link from the Bundle's links.
-
-        Args:
-            bundle (Bundle): The FHIR Bundle containing pagination links.
-
-        Returns:
-            Optional[str]: The URL of the next page if available, None otherwise.
-        """
-        if not bundle.link:
-            return None
-
-        for link in bundle.link:
-            if link.relation == "next":
-                return link.url
-        return None
-
-    def _sanitize_next_link(self, next_link: str) -> str:
-        """
-        Sanitize the `next` link to ensure it is safe to use.
-
-        Args:
-            next_link (str): The raw `next` link URL.
-
-        Returns:
-            str: The sanitized URL.
-
-        Raises:
-            ValueError: If the URL scheme or domain is invalid.
-        """
-        parsed_url = urllib.parse.urlparse(next_link)
-
-        # Validate scheme and netloc (domain)
-        if parsed_url.scheme not in ["http", "https"]:
-            raise ValueError("Invalid URL scheme in `next` link.")
-        if not parsed_url.netloc:
-            raise ValueError("Invalid URL domain in `next` link.")
-
-        # Additional sanitization if necessary, e.g., removing dangerous query parameters
-        query_params = urllib.parse.parse_qs(parsed_url.query)
-        sanitized_query = {k: v for k, v in query_params.items()}
-
-        # Rebuild the sanitized URL
-        sanitized_url = urllib.parse.urlunparse(
-            (
-                parsed_url.scheme,
-                parsed_url.netloc,
-                parsed_url.path,
-                parsed_url.params,
-                urllib.parse.urlencode(sanitized_query, doseq=True),
-                parsed_url.fragment,
-            )
-        )
-
-        return sanitized_url
-
-    def _execute_pagination_request(self, sanitized_url: str) -> Optional[Bundle]:
-        """
-        Execute the request to retrieve the next page using the sanitized URL.
-
-        Args:
-            sanitized_url (str): The sanitized URL to fetch the next page.
-
-        Returns:
-            Optional[Bundle]: The next page of results as a FHIR Bundle, or None.
-
-        Raises:
-            HTTPError: If the request fails due to network issues or server errors.
-        """
-        try:
-            # Use requests.get directly to make the HTTP request
-            response = requests.get(sanitized_url)
-            response.raise_for_status()
-            next_bundle_data = response.json()
-            next_bundle = Bundle(next_bundle_data)
-
-            return next_bundle
-
-        except requests.exceptions.HTTPError as e:
-            # Handle specific HTTP errors as needed, possibly including retry logic
-            raise e
-
-    def iter_pages(self, first_bundle: Bundle) -> Iterable[Bundle]:
-        """
-        Iterator that yields each page of results as a FHIR Bundle.
-
-        Args:
-            first_bundle (Bundle): The first Bundle to start pagination.
-
-        Yields:
-            Bundle: Each page of results as a FHIR Bundle.
-        """
-        bundle = first_bundle
-        while bundle:
-            yield bundle
-            bundle = self._fetch_next_page(bundle)
-

fhirclient/models/bundle.py

@@ -50,7 +50,10 @@ def __init__(self, jsondict=None, strict=True):
         Type `str`. """
 
         super(Bundle, self).__init__(jsondict=jsondict, strict=strict)
-
+
+    def __iter__(self):
+        return iter(self.entry or [])
+
     def elementProperties(self):
         js = super(Bundle, self).elementProperties()
         js.extend([

fhirclient/models/fhirsearch.py

@@ -5,14 +5,22 @@
 #  2014, SMART Health IT.
 
 import logging
+import warnings
+from typing import Iterator, TYPE_CHECKING
+
 
 from . import fhirreference
+from .._utils import iter_pages
 
 try:
     from urllib import quote_plus
 except Exception as e:
     from urllib.parse import quote_plus
 
+if TYPE_CHECKING:
+    from fhirclient.models.resource import Resource
+    from fhirclient.models.bundle import Bundle
+
 logger = logging.getLogger(__name__)
 
 
@@ -110,35 +118,64 @@ def include(self, reference_field, reference_model=None, reverse=False):
         self.includes.append((reference_model, reference_field, reverse))
         return self
 
-    def perform(self, server):
+    def perform(self, server) -> 'Bundle':
         """ Construct the search URL and execute it against the given server.
         
         :param server: The server against which to perform the search
         :returns: A Bundle resource
         """
+        # Old method with deprecation warning
+        warnings.warn(
+            "perform() is deprecated and will be removed in a future release. "
+            "Please use perform_iter() instead.",
+            DeprecationWarning,
+        )
+
         if server is None:
             raise Exception("Need a server to perform search")
 
-        from . import bundle
+        from . import bundle as bundle_module
         res = server.request_json(self.construct())
-        bundle = bundle.Bundle(res)
+        bundle = bundle_module.Bundle(res)
         bundle.origin_server = server
         return bundle
-
-    def perform_resources(self, server):
-        """ Performs the search by calling `perform`, then extracts all Bundle
-        entries and returns a list of Resource instances.
+
+    # Use forward references to avoid circular imports
+    def perform_iter(self, server) -> Iterator['Bundle']:
+        """ Perform the search by calling `perform` and return an iterator that yields
+        Bundle instances.
+
+        :param server: The server against which to perform the search
+        :returns: An iterator of Bundle instances
+        """
+        return iter_pages(self.perform(server))
+
+    def perform_resources(self, server) -> 'list[Resource]':
+        """ Performs the search by calling `perform_resources_iter` and returns a list of Resource instances.
         
         :param server: The server against which to perform the search
         :returns: A list of Resource instances
         """
-        bundle = self.perform(server)
-        resources = []
-        if bundle is not None and bundle.entry is not None:
-            for entry in bundle.entry:
-                resources.append(entry.resource)
-
-        return resources
+        # Old method with deprecation warning
+        warnings.warn(
+            "perform_resources() is deprecated and will be removed in a future release. "
+            "Please use perform_resources_iter() instead.",
+            DeprecationWarning,
+        )
+
+        return list(self.perform_resources_iter(server))
+
+    # Use forward references to avoid circular imports
+    def perform_resources_iter(self, server) -> Iterator['Resource']:
+        """ Performs the search by calling `perform_iter` and yields Resource instances
+        from each Bundle returned by the search.
+
+        :param server: The server against which to perform the search
+        :returns: An iterator of Resource instances
+        """
+        for bundle in self.perform_iter(server):
+            for entry in bundle:
+                yield entry.resource
 
 
 class FHIRSearchParam(object):