Add prefetch_pages to _get_multiple for overlapped page fetching

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Abel Milash

src/PowerPlatform/Dataverse/data/_odata.py

@@ -840,6 +840,7 @@ def _get_multiple(
         page_size: Optional[int] = None,
         count: bool = False,
         include_annotations: Optional[str] = None,
+        prefetch_pages: int = 0,
     ) -> Iterable[List[Dict[str, Any]]]:
         """Iterate records from an entity set, yielding one page (list of dicts) at a time.
 
@@ -861,6 +862,11 @@ def _get_multiple(
         :type count: ``bool``
         :param include_annotations: OData annotation pattern for the ``Prefer: odata.include-annotations`` header (e.g. ``"*"`` or ``"OData.Community.Display.V1.FormattedValue"``), or ``None``.
         :type include_annotations: ``str`` | ``None``
+        :param prefetch_pages: Number of pages to pre-fetch ahead of the caller. ``0`` (default) is
+            fully sequential. ``1`` submits the next HTTP request immediately after receiving the
+            current page, overlapping network I/O with the caller's processing time. Values above
+            ``1`` are capped at ``1``.
+        :type prefetch_pages: ``int``
 
         :return: Iterator yielding pages (each page is a ``list`` of record dicts).
         :rtype: ``Iterable[list[dict[str, Any]]]``
@@ -905,21 +911,38 @@ def _do_request(url: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[st
         if count:
             params["$count"] = "true"
 
-        data = _do_request(base_url, params=params)
-        items = data.get("value") if isinstance(data, dict) else None
-        if isinstance(items, list) and items:
-            yield [x for x in items if isinstance(x, dict)]
-
-        next_link = None
-        if isinstance(data, dict):
-            next_link = data.get("@odata.nextLink") or data.get("odata.nextLink")
-
-        while next_link:
-            data = _do_request(next_link)
+        if prefetch_pages <= 0:
+            data = _do_request(base_url, params=params)
             items = data.get("value") if isinstance(data, dict) else None
             if isinstance(items, list) and items:
                 yield [x for x in items if isinstance(x, dict)]
-            next_link = data.get("@odata.nextLink") or data.get("odata.nextLink") if isinstance(data, dict) else None
+
+            next_link = None
+            if isinstance(data, dict):
+                next_link = data.get("@odata.nextLink") or data.get("odata.nextLink")
+
+            while next_link:
+                data = _do_request(next_link)
+                items = data.get("value") if isinstance(data, dict) else None
+                if isinstance(items, list) and items:
+                    yield [x for x in items if isinstance(x, dict)]
+                next_link = data.get("@odata.nextLink") or data.get("odata.nextLink") if isinstance(data, dict) else None
+        else:
+            # Submit the next request before yielding the current page so
+            # network I/O overlaps with the caller's processing time.
+            ctx = copy_context()
+            executor = ThreadPoolExecutor(max_workers=1)
+            try:
+                pending = executor.submit(ctx.run, _do_request, base_url, params=params)
+                while pending is not None:
+                    data = pending.result()
+                    items = data.get("value") if isinstance(data, dict) else None
+                    next_link = (data.get("@odata.nextLink") or data.get("odata.nextLink")) if isinstance(data, dict) else None
+                    pending = executor.submit(ctx.run, _do_request, next_link) if next_link else None
+                    if isinstance(items, list) and items:
+                        yield [x for x in items if isinstance(x, dict)]
+            finally:
+                executor.shutdown(wait=False, cancel_futures=True)
 
     # --------------------------- SQL Custom API -------------------------
     def _query_sql(self, sql: str) -> list[dict[str, Any]]:

src/PowerPlatform/Dataverse/operations/records.py

@@ -301,6 +301,7 @@ def get(
         page_size: Optional[int] = None,
         count: bool = False,
         include_annotations: Optional[str] = None,
+        prefetch_pages: int = 0,
     ) -> Iterable[List[Record]]:
         """Fetch multiple records from a Dataverse table with pagination.
 
@@ -338,6 +339,11 @@ def get(
             ``Prefer: odata.include-annotations`` header (e.g. ``"*"`` or
             ``"OData.Community.Display.V1.FormattedValue"``), or ``None``.
         :type include_annotations: :class:`str` or None
+        :param prefetch_pages: When ``1``, the next page is fetched in a
+            background thread while the caller processes the current page,
+            overlapping network I/O with processing. ``0`` (default) is fully
+            sequential. Values above ``1`` are capped at ``1``.
+        :type prefetch_pages: :class:`int`
 
         :return: Generator yielding pages, where each page is a list of
             :class:`~PowerPlatform.Dataverse.models.record.Record` objects.
@@ -370,6 +376,7 @@ def get(
         page_size: Optional[int] = None,
         count: bool = False,
         include_annotations: Optional[str] = None,
+        prefetch_pages: int = 0,
     ) -> Union[Record, Iterable[List[Record]]]:
         """Fetch a single record by ID, or fetch multiple records with pagination.
 
@@ -485,6 +492,7 @@ def _paged() -> Iterable[List[Record]]:
                     page_size=page_size,
                     count=count,
                     include_annotations=include_annotations,
+                    **({"prefetch_pages": prefetch_pages} if prefetch_pages else {}),
                 ):
                     yield [Record.from_api_response(table, row) for row in page]
 

tests/unit/data/test_odata_internal.py

@@ -1086,6 +1086,81 @@ def test_empty_value_list_yields_nothing(self):
         self.assertEqual(pages, [])
 
 
+class TestGetMultiplePrefetch(unittest.TestCase):
+    """Behavioral tests for _get_multiple with prefetch_pages=1."""
+
+    def setUp(self):
+        self.od = _make_odata_client()
+        self.od._entity_set_from_schema_name = MagicMock(return_value="accounts")
+
+    def _two_page_responses(self):
+        page1 = _mock_response(
+            json_data={"value": [{"accountid": "id-1"}], "@odata.nextLink": "https://example.crm.dynamics.com/api/data/v9.2/accounts?$skiptoken=2"},
+            text="...",
+        )
+        page2 = _mock_response(
+            json_data={"value": [{"accountid": "id-2"}]},
+            text="...",
+        )
+        self.od._request.side_effect = [page1, page2]
+
+    def test_prefetch_returns_same_pages_as_sequential(self):
+        """prefetch_pages=1 yields the same records in the same order as sequential mode."""
+        self._two_page_responses()
+        pages = list(self.od._get_multiple("account", prefetch_pages=1))
+        self.assertEqual(len(pages), 2)
+        self.assertEqual(pages[0][0]["accountid"], "id-1")
+        self.assertEqual(pages[1][0]["accountid"], "id-2")
+
+    def test_prefetch_values_above_one_capped_at_one(self):
+        """prefetch_pages=2 (or any value > 1) behaves identically to prefetch_pages=1."""
+        self._two_page_responses()
+        pages = list(self.od._get_multiple("account", prefetch_pages=2))
+        self.assertEqual(len(pages), 2)
+        self.assertEqual(pages[0][0]["accountid"], "id-1")
+        self.assertEqual(pages[1][0]["accountid"], "id-2")
+        # Both pages should have been fetched (max_workers=1 means sequential requests)
+        self.assertEqual(self.od._request.call_count, 2)
+
+    def test_prefetch_exception_propagates_to_caller(self):
+        """An exception raised by the background request surfaces when the caller consumes the next page."""
+        page1 = _mock_response(
+            json_data={"value": [{"accountid": "id-1"}], "@odata.nextLink": "https://example.crm.dynamics.com/api/data/v9.2/accounts?$skiptoken=2"},
+            text="...",
+        )
+        self.od._request.side_effect = [page1, RuntimeError("network failure")]
+        gen = self.od._get_multiple("account", prefetch_pages=1)
+        first = next(gen)
+        self.assertEqual(first[0]["accountid"], "id-1")
+        with self.assertRaises(RuntimeError):
+            next(gen)
+
+    def test_prefetch_early_close_does_not_hang(self):
+        """Closing the generator early (break) does not deadlock or raise an error."""
+        page1 = _mock_response(
+            json_data={"value": [{"accountid": "id-1"}], "@odata.nextLink": "https://example.crm.dynamics.com/api/data/v9.2/accounts?$skiptoken=2"},
+            text="...",
+        )
+        page2 = _mock_response(
+            json_data={"value": [{"accountid": "id-2"}]},
+            text="...",
+        )
+        self.od._request.side_effect = [page1, page2]
+        gen = self.od._get_multiple("account", prefetch_pages=1)
+        first = next(gen)
+        self.assertEqual(first[0]["accountid"], "id-1")
+        gen.close()  # should not raise or hang
+
+    def test_prefetch_single_page_no_nextlink(self):
+        """prefetch_pages=1 with a single page (no nextLink) yields exactly one page."""
+        data = {"value": [{"accountid": "id-1"}]}
+        self.od._request.return_value = _mock_response(json_data=data, text=str(data))
+        pages = list(self.od._get_multiple("account", prefetch_pages=1))
+        self.assertEqual(len(pages), 1)
+        self.assertEqual(pages[0][0]["accountid"], "id-1")
+        self.od._request.assert_called_once()
+
+
 class TestQuerySql(unittest.TestCase):
     """Unit tests for _ODataClient._query_sql."""