diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index aa3bbbc5e..5e906a3c0 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 cd91fe758..491f907d2 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 e29f863c8..6240210fe 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,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 diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 21a92cc3b..1ce82a2c7 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 aa546c73a..119648796 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 848ec1e9b..9dc0c0a39 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,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"1", + [{"root": {"item": "1"}}], + id="simple_element", + ), + pytest.param( + b'Book', + [{"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("1"), id="gzipped"), + pytest.param(b"1", 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 = ( + 'Jörg'.encode( + "iso-8859-1" + ) + ) + + assert list(XmlParser(encoding="iso-8859-1").parse(BytesIO(xml_content))) == [ + {"root": {"name": "Jörg"}} + ] 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 21c99adc7..e9f42cf17 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,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'42' + 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"]}}