fix(declarative): support gzip-compressed XML decoding

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,30 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
             yield row
 
 
+@dataclass
+class XmlParser(Parser):
+    """Parses an XML document read from a byte stream into a single record.
+
+    Unlike `XmlDecoder`, which needs a whole `requests.Response`, this parser reads from a
+    byte stream, so it can be nested inside `GzipParser` or `ZipfileDecoder` to handle
+    compressed XML payloads. It is not an incremental parser: the whole document is
+    materialized in memory and emitted as one record.
+
+    Records keep the same shape as `airbyte_cdk.sources.declarative.decoders.XmlDecoder`:
+    attributes are prefixed with `@` and the text content of an element carrying attributes
+    is exposed under `#text`. XML namespace declarations are not supported.
+    """
+
+    encoding: Optional[str] = None
+
+    def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
+        try:
+            yield xmltodict.parse(data, encoding=self.encoding)
+        except ExpatError as exc:
+            logger.warning(f"Response cannot be parsed from XML: {exc}")
+            yield {}
+
+
 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,52 @@ 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(
+            b"<root><item>1</item></root>",
+            [{"root": {"item": "1"}}],
+            id="simple_element",
+        ),
+        pytest.param(
+            b'<root><item id="1" category="books">Book</item></root>',
+            [{"root": {"item": {"@id": "1", "@category": "books", "#text": "Book"}}}],
+            id="attributes_and_text_content",
+        ),
+        pytest.param(
+            b"not xml",
+            [{}],
+            id="malformed_document_yields_empty_record",
+        ),
+    ],
+)
+def test_xml_parser(xml_content: bytes, expected: List[dict]):
+    assert list(XmlParser().parse(BytesIO(xml_content))) == expected
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        pytest.param(compress_with_gzip("<root><item>1</item></root>"), id="gzipped"),
+        pytest.param(b"<root><item>1</item></root>", id="not_gzipped"),
+    ],
+)
+def test_gzip_parser_composes_with_xml_parser(payload: bytes):
+    parser = GzipParser(inner_parser=XmlParser())
+
+    assert list(parser.parse(BytesIO(payload))) == [{"root": {"item": "1"}}]
+
+
+def test_xml_parser_honors_encoding():
+    xml_content = (
+        '<?xml version="1.0" encoding="iso-8859-1"?><root><name>Jörg</name></root>'.encode(
+            "iso-8859-1"
+        )
+    )
+
+    assert list(XmlParser(encoding="iso-8859-1").parse(BytesIO(xml_content))) == [
+        {"root": {"name": "Jörg"}}
+    ]

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,41 @@ 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",
+    [
+        pytest.param({"Content-Encoding": "gzip"}, id="content_encoding_gzip"),
+        pytest.param({"Content-Type": "application/gzip"}, id="content_type_gzip"),
+        pytest.param({"Content-Type": "application/xml"}, id="no_content_encoding_header"),
+    ],
+)
+@pytest.mark.parametrize("emit_connector_builder_messages", [False, True])
+def test_create_gzip_decoder_handles_compressed_xml_response(
+    headers: Mapping[str, str], emit_connector_builder_messages: bool
+):
+    xml_data = b'<orders><order id="1"><units>42</units></order></orders>'
+    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)) == [{"orders": {"order": {"@id": "1", "units": "42"}}}]
+
+
 def test_create_component_type_mismatch():
     manifest = {"check": {"type": "MismatchType", "stream_names": ["list_stream"]}}