Repository: meltano/sdk

Issue:
feat: Support REST paginators that continue after receiving an empty page
### Feature scope

Taps (catalog, state, tests, etc.)

### Description

The break is happening in

https://github.com/meltano/sdk/blob/65575b5f72c9c3a69886a3805a07a61b7a72eb90/singer_sdk/streams/rest.py#L462-L475

but we could do something like

```python
                try:
                    first_record = next(records)
                except StopIteration:
                    if paginator.advance_if_empty(resp):
                        paginator.advance(resp)
                        continue

                    self.logger.info(
                        "Pagination stopped after %d pages because no records were "
                        "found in the last response",
                        pages,
                    )
                    break
```

## Related

- https://github.com/meltano/sdk/issues/2318

Discussion:

Solution patch:
diff --git a/singer_sdk/pagination.py b/singer_sdk/pagination.py
index 9dc630a3..5aa572e0 100644
--- a/singer_sdk/pagination.py
+++ b/singer_sdk/pagination.py
@@ -93,6 +93,20 @@ class BaseAPIPaginator(t.Generic[TPageToken], metaclass=ABCMeta):
         """
         return str(self)
 
+    def continue_if_empty(self, response: requests.Response) -> bool:  # noqa: ARG002, PLR6301
+        """Check if pagination should continue even if the response is empty.
+
+        Override this method to keep the pagination going even if the page retrieved
+        from the API is empty but the next one may have records.
+
+        Args:
+            response: API response object.
+
+        Returns:
+            Boolean flag used to indicate if the endpoint has more pages.
+        """
+        return False
+
     def advance(self, response: requests.Response) -> None:
         """Get a new page value and advance the current one.
 
diff --git a/singer_sdk/streams/rest.py b/singer_sdk/streams/rest.py
index 88110636..c437d58f 100644
--- a/singer_sdk/streams/rest.py
+++ b/singer_sdk/streams/rest.py
@@ -462,6 +462,10 @@ class _HTTPStream(Stream, t.Generic[_TToken], metaclass=abc.ABCMeta):  # noqa: P
                 try:
                     first_record = next(records)
                 except StopIteration:
+                    if paginator.continue_if_empty(resp):
+                        paginator.advance(resp)
+                        continue
+
                     self.logger.info(
                         "Pagination stopped after %d pages because no records were "
                         "found in the last response",


Test patch:
diff --git a/tests/core/rest/test_pagination.py b/tests/core/rest/test_pagination.py
index 7425e22c..73fdb107 100644
--- a/tests/core/rest/test_pagination.py
+++ b/tests/core/rest/test_pagination.py
@@ -346,11 +346,9 @@ def test_break_pagination(tap: Tap, caplog: pytest.LogCaptureFixture):
         name = "my-api-stream"
         url_base = "https://my.api.test"
         path = "/path/to/resource"
+        records_jsonpath = "$.data[*]"
         schema = {"type": "object", "properties": {"id": {"type": "integer"}}}  # noqa: RUF012
 
-        def parse_response(self, response: Response) -> t.Iterable[dict]:
-            return response.json()
-
         def get_new_paginator(self) -> BasePageNumberPaginator:
             return BasePageNumberPaginator(1)
 
@@ -376,14 +374,13 @@ def test_break_pagination(tap: Tap, caplog: pytest.LogCaptureFixture):
             query = parse_qs(parsed.query)
 
             if query.get("page", ["1"]) == ["1"]:
-                r._content = json.dumps(
-                    [
-                        {"id": 1},
-                        {"id": 2},
-                    ]
-                ).encode()
+                r._content = json.dumps({"data": [{"id": 1}, {"id": 2}]}).encode()
+            elif query.get("page", ["2"]) == ["2"]:
+                r._content = json.dumps({"data": []}).encode()
+            elif query.get("page", ["3"]) == ["3"]:
+                r._content = json.dumps({"data": [{"id": 3}, {"id": 4}]}).encode()
             else:
-                r._content = json.dumps([]).encode()
+                r._content = json.dumps({"data": []}).encode()
 
             return r
 
@@ -391,9 +388,90 @@ def test_break_pagination(tap: Tap, caplog: pytest.LogCaptureFixture):
 
     records_iter = stream.request_records(context=None)
 
-    next(records_iter)
-    next(records_iter)
+    assert next(records_iter) == {"id": 1}
+    assert next(records_iter) == {"id": 2}
+
+    with pytest.raises(StopIteration):
+        assert next(records_iter) == {"id": 3}
+
+    with pytest.raises(StopIteration):
+        assert next(records_iter) == {"id": 4}
+
     with pytest.raises(StopIteration):
         next(records_iter)
 
     assert "Pagination stopped after 1 pages" in caplog.text
+
+
+def test_continue_if_empty(tap: Tap):
+    class _TestPaginator(BasePageNumberPaginator):
+        def has_more(self, response: Response) -> bool:
+            return response.json().get("hasMore", False)
+
+        def continue_if_empty(self, response: Response) -> bool:  # noqa: ARG002
+            return True
+
+    class MyAPIStream(RESTStream[int]):
+        """My API stream."""
+
+        name = "my-api-stream"
+        url_base = "https://my.api.test"
+        path = "/path/to/resource"
+        records_jsonpath = "$.data[*]"
+        schema = {"type": "object", "properties": {"id": {"type": "integer"}}}  # noqa: RUF012
+
+        def get_new_paginator(self) -> BasePageNumberPaginator:
+            return _TestPaginator(1)
+
+        def get_url_params(
+            self,
+            context: dict | None,  # noqa: ARG002
+            next_page_token: int | None,
+        ) -> dict[str, t.Any] | str:
+            params = {}
+            if next_page_token:
+                params["page"] = next_page_token
+            return params
+
+        def _request(
+            self,
+            prepared_request: PreparedRequest,
+            context: dict | None,  # noqa: ARG002
+        ) -> Response:
+            r = Response()
+            r.status_code = 200
+
+            parsed = urlparse(prepared_request.url)
+            query = parse_qs(parsed.query)
+
+            if query.get("page", ["1"]) == ["1"]:
+                r._content = json.dumps(
+                    {
+                        "data": [{"id": 1}, {"id": 2}],
+                        "hasMore": True,
+                    }
+                ).encode()
+            elif query.get("page", ["2"]) == ["2"]:
+                r._content = json.dumps({"data": [], "hasMore": True}).encode()
+            elif query.get("page", ["3"]) == ["3"]:
+                r._content = json.dumps(
+                    {
+                        "data": [{"id": 3}, {"id": 4}],
+                        "hasMore": True,
+                    }
+                ).encode()
+            else:
+                r._content = json.dumps({"data": [], "hasMore": False}).encode()
+
+            return r
+
+    stream = MyAPIStream(tap=tap)
+    records_iter = stream.request_records(context=None)
+
+    assert next(records_iter) == {"id": 1}
+    assert next(records_iter) == {"id": 2}
+    assert next(records_iter) == {"id": 3}
+    assert next(records_iter) == {"id": 4}
+
+    with pytest.raises(StopIteration):
+        next(records_iter)