diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index aa3bbbc5e0..5e906a3c0e 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/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: ',')." diff --git a/airbyte_cdk/sources/declarative/decoders/__init__.py b/airbyte_cdk/sources/declarative/decoders/__init__.py index cd91fe758e..491f907d21 100644 --- a/airbyte_cdk/sources/declarative/decoders/__init__.py +++ b/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", ] diff --git a/airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py b/airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py index e29f863c88..7ea5a41e1e 100644 --- a/airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py +++ b/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 diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 21a92cc3be..1ce82a2c75 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/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", diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index aa546c73ab..119648796d 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/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}") diff --git a/unit_tests/sources/declarative/decoders/test_composite_decoder.py b/unit_tests/sources/declarative/decoders/test_composite_decoder.py index 848ec1e9bc..dae3d89228 100644 --- a/unit_tests/sources/declarative/decoders/test_composite_decoder.py +++ b/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( + "Book", + [{"root": {"item": {"@id": "1", "name": "Book"}}}], + id="single-element", + ), + pytest.param( + "ab", + [{"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 = "Book" + 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 = "Book" + 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"}}}] diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 21c99adc71..e580f8b7d0 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/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"Book" + 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"]}}