feat(low-code): stream paginated JSON items with opt-in spool_to_disk and retry-safe downloads

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -107,6 +107,17 @@
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 
+def _decoder_spools_to_disk(decoder_definition: Optional[Dict[str, Any]]) -> bool:
+    """Whether a resolved decoder manifest dict enables spool_to_disk (GzipDecoder recurses)."""
+    if not isinstance(decoder_definition, dict):
+        return False
+    if decoder_definition.get("type") == "JsonItemsDecoder":
+        return bool(decoder_definition.get("spool_to_disk"))
+    if decoder_definition.get("type") == "GzipDecoder":
+        return _decoder_spools_to_disk(decoder_definition.get("decoder"))
+    return False
+
+
 @dataclass
 class TestLimits:
     __test__: ClassVar[bool] = False  # Tell Pytest this is not a Pytest class, despite its name
@@ -530,8 +541,12 @@ def _initialize_cache_for_parent_streams(
         """
         parent_streams = set()
 
-        def _set_cache_if_not_disabled(requester: Dict[str, Any]) -> None:
+        def _set_cache_if_not_disabled(retriever: Dict[str, Any]) -> None:
             """Set use_cache to True only if not explicitly disabled."""
+            if _decoder_spools_to_disk(retriever.get("decoder")):
+                # requests_cache consumes the whole body, leaving nothing to spool
+                return
+            requester = retriever["requester"]
             if requester.get("use_cache") is not False:
                 requester["use_cache"] = True
 
@@ -542,19 +557,19 @@ def update_with_cache_parent_configs(
                 parent_streams.add(parent_config["stream"]["name"])
                 if parent_config["stream"]["type"] == "StateDelegatingStream":
                     _set_cache_if_not_disabled(
-                        parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"]
+                        parent_config["stream"]["full_refresh_stream"]["retriever"]
                     )
                     _set_cache_if_not_disabled(
-                        parent_config["stream"]["incremental_stream"]["retriever"]["requester"]
+                        parent_config["stream"]["incremental_stream"]["retriever"]
                     )
                 else:
-                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"]["requester"])
+                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"])
 
         for stream_config in stream_configs:
             if stream_config.get("incremental_sync", {}).get("parent_stream"):
                 parent_streams.add(stream_config["incremental_sync"]["parent_stream"]["name"])
                 _set_cache_if_not_disabled(
-                    stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"]
+                    stream_config["incremental_sync"]["parent_stream"]["retriever"]
                 )
 
             elif stream_config.get("retriever", {}).get("partition_router", {}):
@@ -581,14 +596,10 @@ def update_with_cache_parent_configs(
         for stream_config in stream_configs:
             if stream_config["name"] in parent_streams:
                 if stream_config["type"] == "StateDelegatingStream":
-                    _set_cache_if_not_disabled(
-                        stream_config["full_refresh_stream"]["retriever"]["requester"]
-                    )
-                    _set_cache_if_not_disabled(
-                        stream_config["incremental_stream"]["retriever"]["requester"]
-                    )
+                    _set_cache_if_not_disabled(stream_config["full_refresh_stream"]["retriever"])
+                    _set_cache_if_not_disabled(stream_config["incremental_stream"]["retriever"])
                 else:
-                    _set_cache_if_not_disabled(stream_config["retriever"]["requester"])
+                    _set_cache_if_not_disabled(stream_config["retriever"])
         return stream_configs
 
     def spec(self, logger: logging.Logger) -> ConnectorSpecification:

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2956,6 +2956,16 @@ definitions:
         description: Text encoding used to decode the streamed bytes before JSON parsing.
         type: string
         default: utf-8
+      spool_to_disk:
+        title: Spool Response To Disk
+        description: >-
+          Download the whole response body to an anonymous temporary file before parsing it, instead of
+          parsing straight from the network socket. The download happens inside the request retry logic, so
+          a connection that breaks mid-body is retried exactly like a failed request and no record of the
+          page is emitted before the full page is on disk. Costs the decoded size of one page in temporary
+          disk per in-flight request; memory stays bounded. Recommended for very large pages consumed slowly.
+        type: boolean
+        default: false
   JsonlDecoder:
     title: JSON Lines
     description: Select 'JSON Lines' if the response consists of JSON objects separated by new lines ('\n') in JSONL format.

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -8,9 +8,10 @@
 import io
 import json
 import logging
+from collections.abc import Mapping
 from dataclasses import dataclass
 from io import BufferedIOBase, TextIOWrapper
-from typing import Any, List, Optional
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple
 
 import ijson
 import orjson
@@ -25,6 +26,7 @@
     PARSERS_TYPE,
     Parser,
 )
+from airbyte_cdk.sources.streams.http.streamed_response import set_document_remainder
 from airbyte_cdk.utils import AirbyteTracedException
 
 logger = logging.getLogger("airbyte")
@@ -63,7 +65,7 @@ def readinto(self, buffer: Buffer) -> int:
 class GzipParser(Parser):
     inner_parser: Parser
 
-    def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
+    def parse(self, data: BufferedIOBase, **kwargs: Any) -> PARSER_OUTPUT_TYPE:
         """Decompress gzipped data or pass uncompressed data through unchanged.
 
         Args:
@@ -82,9 +84,9 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
 
         if prefix == b"\x1f\x8b":
             with gzip.GzipFile(fileobj=prefixed_data, mode="rb") as gzipobj:
-                yield from self.inner_parser.parse(gzipobj)
+                yield from self.inner_parser.parse(gzipobj, **kwargs)
         else:
-            yield from self.inner_parser.parse(prefixed_data)
+            yield from self.inner_parser.parse(prefixed_data, **kwargs)
 
 
 @dataclass
@@ -158,6 +160,21 @@ def read(self, size: int = -1) -> bytes:
         return self._decoder.decode(chunk, final=not chunk).encode("utf-8")
 
 
+_INTEGER_OVERFLOW_MESSAGE = "integer overflow"
+
+
+def _resolve_items_path(document: Any, items_path: str) -> Tuple[Any, str, List[Any]]:
+    """Walks an ijson dotted `items_path` and returns (parent, last_key, items_list)."""
+    *parents, key = items_path.split(".")
+    node = document
+    for part in parents:
+        node = node[part]
+    items = node[key]
+    if not isinstance(items, list):
+        raise ValueError(f"items_path {items_path!r} does not point at a JSON array.")
+    return node, key, items
+
+
 @dataclass
 class JsonItemsParser(Parser):
     """Streaming JSON parser that yields each element of a nested array.
@@ -176,7 +193,12 @@ class JsonItemsParser(Parser):
     items_path: str = ""
     encoding: Optional[str] = "utf-8"
 
-    def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
+    def parse(
+        self,
+        data: BufferedIOBase,
+        *,
+        on_document_remainder: Optional[Callable[[Any], None]] = None,
+    ) -> PARSER_OUTPUT_TYPE:
         if not self.items_path:
             raise ValueError("JsonItemsParser requires a non-empty items_path.")
         if self.encoding and codecs.lookup(self.encoding).name != "utf-8":
@@ -185,12 +207,50 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
             # so ijson keeps using its fast byte backend rather than a (deprecated) text
             # stream. The recoder decodes lazily in chunks, preserving bounded memory.
             data = _Utf8Recoder(data, self.encoding)  # type: ignore[assignment]
+        source = data if not isinstance(data, _Utf8Recoder) else data._stream
         # ijson auto-selects the best available backend (yajl2_c when present)
         # and reads from `data` lazily — it does not call `.read()` on the
         # whole stream up front.
         # use_float=True yields floats for non-integer numbers instead of Decimal, matching
         # json.loads/orjson behavior so downstream JSON serialization doesn't choke on Decimal.
-        yield from ijson.items(data, f"{self.items_path}.item", use_float=True)
+        items_prefix = f"{self.items_path}.item"
+        yielded = 0
+        try:
+            if on_document_remainder is None:
+                for item in ijson.items(data, items_prefix, use_float=True):
+                    yield item
+                    yielded += 1
+                return
+            builder = ijson.common.ObjectBuilder()
+
+            def events() -> Generator[Tuple[str, str, Any], None, None]:
+                for prefix, event, value in ijson.parse(data, use_float=True):
+                    if not (prefix == items_prefix or prefix.startswith(items_prefix + ".")):
+                        builder.event(event, value)
+                    yield prefix, event, value
+
+            for item in ijson.items(events(), items_prefix):
+                yield item
+                yielded += 1
+            on_document_remainder(builder.value)
+        except ijson.common.IncompleteJSONError as exc:
+            if _INTEGER_OVERFLOW_MESSAGE not in str(exc) or not source.seekable():
+                raise
+            # yajl (ijson's C backend) rejects integers outside int64. The spooled body is
+            # rewindable, so fall back to a full (non-streaming) orjson parse of the same bytes,
+            # as JsonParser does, and skip the items this response already emitted.
+            logger.warning(
+                "JsonItemsParser: integer outside int64 in response; re-parsing spooled body "
+                "with orjson after %d items",
+                yielded,
+            )
+            source.seek(0)
+            document = orjson.loads(source.read().decode(self.encoding or "utf-8"))
+            parent, key, items = _resolve_items_path(document, self.items_path)
+            yield from items[yielded:]
+            if on_document_remainder is not None:
+                parent[key] = []
+                on_document_remainder(document)
 
 
 @dataclass
@@ -222,6 +282,14 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
             yield row
 
 
+def _parser_supports_document_remainder(parser: Parser) -> bool:
+    if isinstance(parser, JsonItemsParser):
+        return True
+    if isinstance(parser, GzipParser):
+        return _parser_supports_document_remainder(parser.inner_parser)
+    return False
+
+
 class CompositeRawDecoder(Decoder):
     """
     Decoder strategy to transform a requests.Response into a PARSER_OUTPUT_TYPE
@@ -242,6 +310,7 @@ def __init__(
         parser: Parser,
         stream_response: bool = True,
         parsers_by_header: PARSERS_BY_HEADER_TYPE = None,
+        spool_response: bool = False,
     ) -> None:
         # since we moved from using `dataclass` to `__init__` method,
         # we need to keep using the `parser` to be able to resolve the depenencies
@@ -250,6 +319,8 @@ def __init__(
 
         self._parsers_by_header = parsers_by_header if parsers_by_header else {}
         self._stream_response = stream_response
+        self._spool_response = spool_response
+        self._capture_document_remainder = False
 
     @classmethod
     def by_headers(
@@ -275,23 +346,47 @@ def by_headers(
                 parsers_by_header[header] = {header_value: parser for header_value in header_values}
         return cls(fallback_parser, stream_response, parsers_by_header)
 
+    def enable_document_remainder_capture(self) -> None:
+        """Opt in to capturing the non-items document fields for pagination."""
+        self._capture_document_remainder = True
+
     def is_stream_response(self) -> bool:
         return self._stream_response
 
+    def spools_response(self) -> bool:
+        return self._spool_response
+
     def decode(self, response: requests.Response) -> DECODER_OUTPUT_TYPE:
         parser = self._select_parser(response)
+
+        parse_kwargs: Dict[str, Any] = {}
+        if self._capture_document_remainder and _parser_supports_document_remainder(parser):
+
+            def on_document_remainder(doc: Any) -> None:
+                set_document_remainder(response, doc)
+                logger.debug(
+                    "Captured pagination document remainder",
+                    extra={
+                        "keys": list(doc.keys()) if isinstance(doc, Mapping) else type(doc).__name__
+                    },
+                )
+
+            parse_kwargs["on_document_remainder"] = on_document_remainder
         if self.is_stream_response():
             # urllib mentions that some interfaces don't play nice with auto_close
             # More info here: https://urllib3.readthedocs.io/en/stable/user-guide.html#using-io-wrappers-with-response-content
             # We have indeed observed some issues with CSV parsing.
             # Hence, we will manage the closing of the file ourselves until we find a better solution.
             response.raw.auto_close = False
+            if hasattr(response.raw, "decode_content"):
+                response.raw.decode_content = True
             yield from parser.parse(
                 data=response.raw,  # type: ignore[arg-type]
+                **parse_kwargs,
             )
             response.raw.close()
         else:
-            yield from parser.parse(data=io.BytesIO(response.content))
+            yield from parser.parse(data=io.BytesIO(response.content), **parse_kwargs)
 
     def _select_parser(self, response: requests.Response) -> Parser:
         """

airbyte_cdk/sources/declarative/decoders/decoder.py

@@ -23,6 +23,12 @@ def is_stream_response(self) -> bool:
         Set to True if you'd like to use stream=True option in http requester
         """
 
+    def spools_response(self) -> bool:
+        """
+        Set to True if the full response body should be copied to disk before decoding.
+        """
+        return False
+
     @abstractmethod
     def decode(self, response: requests.Response) -> DECODER_OUTPUT_TYPE:
         """

airbyte_cdk/sources/declarative/decoders/pagination_decoder_decorator.py

@@ -9,6 +9,7 @@
 import requests
 
 from airbyte_cdk.sources.declarative.decoders import Decoder
+from airbyte_cdk.sources.streams.http.streamed_response import get_document_remainder
 
 logger = logging.getLogger("airbyte")
 
@@ -33,7 +34,13 @@ def decode(
         self, response: requests.Response
     ) -> Generator[MutableMapping[str, Any], None, None]:
         if self._decoder.is_stream_response():
-            logger.warning("Response is streamed and therefore will not be decoded for pagination.")
-            yield {}
+            remainder = get_document_remainder(response)
+            if remainder is not None:
+                yield remainder
+            else:
+                logger.warning(
+                    "Response is streamed and therefore will not be decoded for pagination."
+                )
+                yield {}
         else:
             yield from self._decoder.decode(response)

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -846,6 +846,11 @@ class JsonItemsDecoder(BaseModel):
         description="The character encoding of the JSON data. Defaults to UTF-8.",
         title="Encoding",
     )
+    spool_to_disk: Optional[bool] = Field(
+        False,
+        description="Download the whole response body to an anonymous temporary file before parsing it, instead of parsing straight from the network socket. The download happens inside the request retry logic, so a connection that breaks mid-body is retried exactly like a failed request and no record of the page is emitted before the full page is on disk. Costs the decoded size of one page in temporary disk per in-flight request; memory stays bounded. Recommended for very large pages consumed slowly.",
+        title="Spool Response To Disk",
+    )
 
 
 class JsonlDecoder(BaseModel):