Add pagination functionality to FHIRClient and related unit tests

fhirclient/client.py

@@ -1,4 +1,7 @@
 import logging
+import urllib
+
+import requests
 
 from .server import FHIRServer, FHIRUnauthorizedException, FHIRNotFoundException
 
@@ -240,3 +243,97 @@ def from_state(self, state):
     def save_state (self):
         self._save_func(self.state)
 
+    # MARK: Pagination
+    def fetch_next_page(self, 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:
+            Bundle: The next page of results as a FHIR Bundle.
+        """
+        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):
+        """
+        Extract the `next` link from the Bundle's links.
+
+        Args:
+            bundle (Bundle): The FHIR Bundle containing pagination links.
+
+        Returns:
+            str: The URL of the next page if available, None otherwise.
+        """
+        for link in bundle["link"]:
+            if link["relation"] == "next":
+                return link["url"]
+        return None
+
+    def sanitize_next_link(self, next_link):
+        """
+        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):
+        """
+        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:
+            Bundle: The next page of results as a FHIR Bundle.
+
+        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()
+            return response.json()
+        except requests.exceptions.HTTPError as e:
+            # Handle specific HTTP errors as needed, possibly including retry logic
+            raise e
+
+

tests/client_pagination_test.py

@@ -0,0 +1,129 @@
+import unittest
+from unittest.mock import patch, MagicMock
+
+import requests
+from fhirclient.client import FHIRClient
+
+
+class TestFHIRClientPagination(unittest.TestCase):
+    def setUp(self) -> None:
+        state = {
+            "app_id": "AppID",
+            "app_secret": "AppSecret",
+            "scope": "user/*.read",
+            "redirect": "http://test.invalid/redirect",
+            "patient_id": "PatientID",
+            "server": {
+                "base_uri": "http://test.invalid/",
+                "auth_type": "none",
+                "auth": {
+                    "app_id": "AppId",
+                },
+            },
+            "launch_token": "LaunchToken",
+            "launch_context": {
+                "encounter": "EncounterID",
+                "patient": "PatientID",
+            },
+            "jwt_token": "JwtToken",
+        }
+
+        # Confirm round trip
+        self.client = FHIRClient(state=state)
+
+        self.bundle = {
+            "link": [
+                {"relation": "self", "url": "http://example.com/fhir/Bundle/1"},
+                {"relation": "next", "url": "http://example.com/fhir/Bundle/2"},
+            ],
+            "entry": [
+                {
+                    "fullUrl": "http://example.com/fhir/Patient/1",
+                    "resource": {
+                        "resourceType": "Patient",
+                        "id": "1",
+                        "name": [{"family": "Doe", "given": ["John"]}],
+                        "gender": "male",
+                        "birthDate": "1980-01-01",
+                    },
+                },
+                {
+                    "fullUrl": "http://example.com/fhir/Patient/2",
+                    "resource": {
+                        "resourceType": "Patient",
+                        "id": "2",
+                        "name": [{"family": "Smith", "given": ["Jane"]}],
+                        "gender": "female",
+                        "birthDate": "1990-05-15",
+                    },
+                },
+            ],
+        }
+
+    def test_get_next_link(self):
+        next_link = self.client.get_next_link(self.bundle)
+        self.assertEqual(next_link, "http://example.com/fhir/Bundle/2")
+
+    def test_get_next_link_no_next(self):
+        bundle_without_next = {"link": [{"relation": "self", "url": "http://example.com/fhir/Bundle/1"}]}
+        next_link = self.client.get_next_link(bundle_without_next)
+        self.assertIsNone(next_link)
+
+    def test_sanitize_next_link_valid(self):
+        next_link = "http://example.com/fhir/Bundle/2?page=2&size=10"
+        sanitized_link = self.client.sanitize_next_link(next_link)
+        self.assertEqual(sanitized_link, next_link)
+
+    def test_sanitize_next_link_invalid_scheme(self):
+        next_link = "ftp://example.com/fhir/Bundle/2?page=2&size=10"
+        with self.assertRaises(ValueError):
+            self.client.sanitize_next_link(next_link)
+
+    def test_sanitize_next_link_invalid_domain(self):
+        next_link = "http:///fhir/Bundle/2?page=2&size=10"
+        with self.assertRaises(ValueError):
+            self.client.sanitize_next_link(next_link)
+
+    @patch("requests.get")
+    def test_execute_pagination_request_success(self, mock_get):
+        mock_response = MagicMock()
+        # Set up the mock to return a specific JSON payload when its json() method is called
+        mock_response.json.return_value = self.bundle
+        mock_response.raise_for_status = MagicMock()
+        mock_get.return_value = mock_response
+
+        next_link = "http://example.com/fhir/Bundle/2"
+        sanitized_link = self.client.sanitize_next_link(next_link)
+        result = self.client.execute_pagination_request(sanitized_link)
+
+        # Check that the result is a dictionary (which is the mocked JSON response)
+        self.assertIsInstance(result, dict)
+        # Ensure that "entry" is a key in the returned dictionary
+        self.assertIn("entry", result)
+        # Assert that requests.get was called exactly once with the sanitized URL
+        mock_get.assert_called_once_with(sanitized_link)
+
+    @patch("requests.get")
+    def test_execute_pagination_request_http_error(self, mock_get):
+        mock_get.side_effect = requests.exceptions.HTTPError("HTTP Error")
+
+        next_link = "http://example.com/fhir/Bundle/2"
+        sanitized_link = self.client.sanitize_next_link(next_link)
+
+        with self.assertRaises(requests.exceptions.HTTPError):
+            self.client.execute_pagination_request(sanitized_link)
+
+    @patch("fhirclient.client.FHIRClient.execute_pagination_request")
+    def test_fetch_next_page(self, mock_execute_request):
+        mock_execute_request.return_value = self.bundle
+
+        result = self.client.fetch_next_page(self.bundle)
+
+        self.assertIsInstance(result, dict)
+        self.assertIn("entry", result)
+        mock_execute_request.assert_called_once()
+
+    def test_fetch_next_page_no_next_link(self):
+        bundle_without_next = {"link": [{"relation": "self", "url": "http://example.com/fhir/Bundle/1"}]}
+        result = self.client.fetch_next_page(bundle_without_next)
+        self.assertIsNone(result)