feat: add Composite Raw Decoder

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2719,9 +2719,44 @@ definitions:
           - "$ref": "#/definitions/IterableDecoder"
           - "$ref": "#/definitions/XmlDecoder"
           - "$ref": "#/definitions/GzipJsonDecoder"
+          - "$ref": "#/definitions/CompositeRawDecoder"
       $parameters:
         type: object
         additionalProperties: true
+  CompositeRawDecoder:
+    description: "(This is experimental, use at your own risk)"
+    type: object
+    required:
+      - parser
+    properties:
+      parser:
+        anyOf:
+          - "$ref": "#/definitions/GzipParser"
+          - "$ref": "#/definitions/JsonLineParser"
+          - "$ref": "#/definitions/CsvParser"
+#    PARSERS
+  GzipParser:
+    type: object
+    properties:
+      inner_parser:
+        anyOf:
+          - "$ref": "#/definitions/JsonLineParser"
+          - "$ref": "#/definitions/CsvParser"
+  JsonLineParser:
+    type: object
+    properties:
+      encoding:
+        type: string
+        default: utf-8
+  CsvParser:
+    type: object
+    properties:
+      encoding:
+        type: string
+        default: utf-8
+      delimiter:
+        type: string
+        default: ","
   AsyncJobStatusMap:
     description: Matches the api job status to Async Job Status.
     type: object

airbyte_cdk/sources/declarative/decoders/composite_decoder.py

@@ -0,0 +1,127 @@
+import gzip
+import json
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Any, Generator, Iterable, MutableMapping, Optional, Protocol
+
+import pandas as pd
+import requests
+from numpy import nan
+
+from airbyte_cdk.sources.declarative.decoders.decoder import Decoder
+
+
+class BufferedDataInput(Protocol):
+    """
+    at least read to _n bytes method should be supported to align with BufferedReader and avoid OOM.
+    """
+
+    def read(self, __n: int) -> bytes: ...
+
+
+@dataclass
+class Parser(ABC):
+    inner_parser: Optional["Parser"] = None
+
+    @abstractmethod
+    def parse(
+        self, data: BufferedDataInput | Iterable, *args, **kwargs
+    ) -> Generator[MutableMapping[str, Any], None, None]:
+        """
+        Parse data and yield dictionaries.
+        """
+        pass
+
+
+@dataclass
+class GzipParser(Parser):
+    def parse(
+        self, data: BufferedDataInput | Iterable, *args, **kwargs
+    ) -> Generator[MutableMapping[str, Any], None, None]:
+        """
+        Decompress gzipped bytes and pass decompressed data to the inner parser.
+        """
+        gzipobj = gzip.GzipFile(fileobj=data)
+        if self.inner_parser:
+            yield from self.inner_parser.parse(gzipobj)
+        else:
+            yield from gzipobj
+
+
+@dataclass
+class JsonLineParser(Parser):
+    encoding: Optional[str] = "utf-8"
+
+    def parse(
+        self, data: BufferedDataInput | Iterable, *args, **kwargs
+    ) -> Generator[MutableMapping[str, Any], None, None]:
+        for line in data:
+            try:
+                yield json.loads(line.decode(self.encoding))
+            except json.JSONDecodeError:
+                # Handle invalid JSON lines gracefully (e.g., log and skip)
+                pass
+
+
+@dataclass
+class CsvParser(Parser):
+    # TODO: add more parameters: see read_csv for more details, e.g.: quotechar, headers,
+    encoding: Optional[str] = "utf-8"
+    delimiter: str = ","
+
+    def parse(
+        self, data: BufferedDataInput | Iterable, *args, **kwargs
+    ) -> Generator[MutableMapping[str, Any], None, None]:
+        """
+        Parse CSV data from decompressed bytes.
+        """
+        reader = pd.read_csv(data, sep=self.delimiter, iterator=True, dtype=object)
+        for chunk in reader:
+            chunk = chunk.replace({nan: None}).to_dict(orient="records")
+            for row in chunk:
+                yield row
+
+
+@dataclass
+class CompositeRawDecoder(Decoder):
+    """
+    Decoder strategy to transform a requests.Response into a Generator[MutableMapping[str, Any], None, None]
+    passed response.raw to parser(s).
+    Note: response.raw is not decoded/decompressed by default.
+    parsers should be instantiated recursively.
+    Example:
+    composite_decoder = CompositeDecoder(parser=GzipParser(inner_parser=JsonLineParser(encoding="iso-8859-1")))
+    """
+
+    parser: Parser
+
+    def is_stream_response(self) -> bool:
+        return True
+
+    def decode(
+        self, response: requests.Response
+    ) -> Generator[MutableMapping[str, Any], None, None]:
+        yield from self.parser.parse(data=response.raw)
+
+
+# Examples how to use
+if __name__ == "__main__":
+    # SIMPLE JSONLINES
+    # composite_decoder = CompositeDecoder(parser=JsonLineParser())
+    # response = requests.get('http://127.0.0.1:5000/jsonlines', stream=True)
+    # for rec in composite_decoder.decode(response):
+    #     print(rec)
+
+    # Gzipped JSONLINES
+    # parser = GzipParser(inner_parser=JsonLineParser(encoding="iso-8859-1"))
+    # composite_decoder = CompositeDecoder(parser=parser)
+    # response = requests.get('http://127.0.0.1:5000/jsonlines', stream=True)
+    # for rec in composite_decoder.decode(response):
+    #     print(rec)
+
+    # Gzipped TSV
+    parser = GzipParser(inner_parser=CsvParser(encoding="iso-8859-1", delimiter="\t"))
+    composite_decoder = CompositeRawDecoder(parser=parser)
+    response = requests.get("http://127.0.0.1:5000/csv", stream=True)
+    for rec in composite_decoder.decode(response):
+        print(rec)

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1109,6 +1109,15 @@ class LegacySessionTokenAuthenticator(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class JsonLineParser(BaseModel):
+    encoding: Optional[str] = "utf-8"
+
+
+class CsvParser(BaseModel):
+    encoding: Optional[str] = "utf-8"
+    delimiter: Optional[str] = ","
+
+
 class AsyncJobStatusMap(BaseModel):
     type: Optional[Literal["AsyncJobStatusMap"]] = None
     running: List[str]
@@ -1488,6 +1497,10 @@ class RecordSelector(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class GzipParser(BaseModel):
+    inner_parser: Optional[Union[JsonLineParser, CsvParser]] = None
+
+
 class Spec(BaseModel):
     type: Literal["Spec"]
     connection_specification: Dict[str, Any] = Field(
@@ -1518,6 +1531,10 @@ class CompositeErrorHandler(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class CompositeRawDecoder(BaseModel):
+    parser: Union[GzipParser, JsonLineParser, CsvParser]
+
+
 class DeclarativeSource1(BaseModel):
     class Config:
         extra = Extra.forbid
@@ -1895,6 +1912,7 @@ class SimpleRetriever(BaseModel):
             IterableDecoder,
             XmlDecoder,
             GzipJsonDecoder,
+            CompositeRawDecoder,
         ]
     ] = Field(
         None,