feat(declarative): support XML inside GzipDecoder and ZipfileDecoder

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3164,6 +3164,7 @@ definitions:
           - "$ref": "#/definitions/JsonDecoder"
           - "$ref": "#/definitions/JsonItemsDecoder"
           - "$ref": "#/definitions/JsonlDecoder"
+          - "$ref": "#/definitions/XmlDecoder"
   ListPartitionRouter:
     title: List Partition Router
     description: A Partition router that specifies a list of attributes where each attribute describes a portion of the complete data set for a stream. During a sync, each value is iterated over and can be used as input to outbound API requests.
@@ -4296,6 +4297,7 @@ definitions:
           - "$ref": "#/definitions/JsonDecoder"
           - "$ref": "#/definitions/JsonItemsDecoder"
           - "$ref": "#/definitions/JsonlDecoder"
+          - "$ref": "#/definitions/XmlDecoder"
   CsvDecoder:
     title: CSV
     description: "Select 'CSV' for response data that is formatted as CSV (comma-separated values). Can specify an encoding (default: 'utf-8') and a delimiter (default: ',')."

airbyte_cdk/sources/declarative/decoders/__init__.py

@@ -7,6 +7,7 @@
     GzipParser,
     JsonParser,
     Parser,
+    XmlParser,
 )
 from airbyte_cdk.sources.declarative.decoders.decoder import Decoder
 from airbyte_cdk.sources.declarative.decoders.json_decoder import (
@@ -29,5 +30,6 @@
     "NoopDecoder",
     "PaginationDecoderDecorator",
     "XmlDecoder",
+    "XmlParser",
     "ZipfileDecoder",
 ]

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -11,10 +11,12 @@
 from dataclasses import dataclass
 from io import BufferedIOBase, TextIOWrapper
 from typing import Any, List, Optional
+from xml.parsers.expat import ExpatError
 
 import ijson
 import orjson
 import requests
+import xmltodict
 from typing_extensions import Buffer
 
 from airbyte_cdk.models import FailureType
@@ -222,6 +224,33 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
             yield row
 
 
+@dataclass
+class XmlParser(Parser):
+    """Parse an XML document into a dictionary.
+
+    Attributes are prefixed with `@` and text content is exposed under `#text` when the
+    element also has attributes, matching the behavior of
+    `airbyte_cdk.sources.declarative.decoders.XmlDecoder`. Unlike that decoder, this parser
+    reads raw bytes, so it can be nested inside compressed parsers such as `GzipParser`.
+    """
+
+    encoding: Optional[str] = "utf-8"
+
+    def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
+        raw_data = data.read()
+        try:
+            body_json = xmltodict.parse(raw_data, encoding=self.encoding)
+        except ExpatError as exc:
+            logger.warning(f"Response cannot be parsed from XML: {exc=}")
+            yield {}
+            return
+
+        if isinstance(body_json, list):
+            yield from body_json
+        else:
+            yield body_json
+
+
 class CompositeRawDecoder(Decoder):
     """
     Decoder strategy to transform a requests.Response into a PARSER_OUTPUT_TYPE

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2384,7 +2384,7 @@ class PaginationReset(BaseModel):
 
 class GzipDecoder(BaseModel):
     type: Literal["GzipDecoder"]
-    decoder: Union[CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder]
+    decoder: Union[CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder, XmlDecoder]
 
 
 class RequestBodyGraphQL(BaseModel):
@@ -2522,7 +2522,9 @@ class Config:
         extra = Extra.allow
 
     type: Literal["ZipfileDecoder"]
-    decoder: Union[CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder] = Field(
+    decoder: Union[
+        CsvDecoder, GzipDecoder, JsonDecoder, JsonItemsDecoder, JsonlDecoder, XmlDecoder
+    ] = Field(
         ...,
         description="Parser to parse the decompressed data from the zipfile(s).",
         title="Parser",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -106,6 +106,7 @@
     JsonLineParser,
     JsonParser,
     Parser,
+    XmlParser,
 )
 from airbyte_cdk.sources.declarative.expanders.record_expander import (
     OnNoRecords,
@@ -2832,13 +2833,13 @@ def _get_parser(model: BaseModel, config: Config) -> Parser:
                 delimiter=model.delimiter,
                 set_values_to_none=model.set_values_to_none,
             )
+        elif isinstance(model, XmlDecoderModel):
+            return XmlParser()
         elif isinstance(model, GzipDecoderModel):
             return GzipParser(
                 inner_parser=ModelToComponentFactory._get_parser(model.decoder, config)
             )
-        elif isinstance(
-            model, (CustomDecoderModel, IterableDecoderModel, XmlDecoderModel, ZipfileDecoderModel)
-        ):
+        elif isinstance(model, (CustomDecoderModel, IterableDecoderModel, ZipfileDecoderModel)):
             raise ValueError(f"Decoder type {model} does not have parser associated to it")
 
         raise ValueError(f"Unknown decoder type {model}")

unit_tests/sources/declarative/decoders/test_composite_decoder.py

@@ -22,6 +22,7 @@
     JsonItemsParser,
     JsonLineParser,
     JsonParser,
+    XmlParser,
 )
 from airbyte_cdk.utils import AirbyteTracedException
 
@@ -565,3 +566,47 @@ def readable(self) -> bool:  # pragma: no cover - interface compliance
     first = next(iterator)
     assert first == {"id": 0, "name": "name-0"}
     assert stream.bytes_read < stream.total_size
+
+
+@pytest.mark.parametrize(
+    "xml_content, expected",
+    [
+        pytest.param(
+            "<root><item id='1'><name>Book</name></item></root>",
+            [{"root": {"item": {"@id": "1", "name": "Book"}}}],
+            id="single-element",
+        ),
+        pytest.param(
+            "<root><item>a</item><item>b</item></root>",
+            [{"root": {"item": ["a", "b"]}}],
+            id="repeated-elements",
+        ),
+        pytest.param("not xml at all", [{}], id="invalid-xml-yields-empty-record"),
+    ],
+)
+def test_xml_parser(xml_content: str, expected: List[dict]):
+    assert list(XmlParser().parse(BytesIO(xml_content.encode("utf-8")))) == expected
+
+
+def test_xml_parser_composes_with_gzip():
+    xml_content = "<root><item id='1'><name>Book</name></item></root>"
+    parser = GzipParser(inner_parser=XmlParser())
+
+    assert list(parser.parse(BytesIO(compress_with_gzip(xml_content)))) == [
+        {"root": {"item": {"@id": "1", "name": "Book"}}}
+    ]
+
+
+def test_composite_raw_decoder_gzip_xml_parser(requests_mock):
+    xml_content = "<root><item id='1'><name>Book</name></item></root>"
+    requests_mock.register_uri(
+        "GET",
+        "https://airbyte.io/",
+        content=compress_with_gzip(xml_content),
+        headers={"Content-Type": "application/gzip"},
+    )
+    response = requests.get("https://airbyte.io/", stream=True)
+
+    decoder = CompositeRawDecoder(parser=GzipParser(inner_parser=XmlParser()))
+
+    assert list(decoder.decode(response)) == [{"root": {"item": {"@id": "1", "name": "Book"}}}]

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -120,6 +120,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     SelectiveAuthenticator,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    XmlDecoder as XmlDecoderModel,
+)
 from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
     ENV_VAR_ALLOW_CUSTOM_CODE,
     INJECTED_MANIFEST,
@@ -335,6 +338,42 @@ def test_create_gzip_decoder_handles_transport_and_content_gzip(
     assert list(decoder.decode(response)) == [{"date": "2026-08-01", "units": "42"}]
 
 
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {"Content-Type": "application/gzip"},
+        {"Content-Type": "application/x-gzip"},
+        {"Content-Type": "application/xml"},
+        {"Content-Encoding": "gzip"},
+    ],
+)
+@pytest.mark.parametrize("emit_connector_builder_messages", [False, True])
+def test_create_gzip_decoder_handles_xml_inner_decoder(
+    headers: Mapping[str, str], emit_connector_builder_messages: bool
+):
+    xml_data = b"<root><item id='1'><name>Book</name></item></root>"
+    response = requests.Response()
+    response.status_code = 200
+    response.headers.update(headers)
+    response.raw = HTTPResponse(
+        body=io.BytesIO(gzip.compress(xml_data)),
+        headers=headers,
+        status=200,
+        preload_content=False,
+        decode_content=False,
+    )
+
+    model = GzipDecoderModel(
+        type="GzipDecoder",
+        decoder=XmlDecoderModel(type="XmlDecoder"),
+    )
+    decoder = ModelToComponentFactory(
+        emit_connector_builder_messages=emit_connector_builder_messages
+    ).create_gzip_decoder(model, {})
+
+    assert list(decoder.decode(response)) == [{"root": {"item": {"@id": "1", "name": "Book"}}}]
+
+
 def test_create_component_type_mismatch():
     manifest = {"check": {"type": "MismatchType", "stream_names": ["list_stream"]}}