From ff96c6b9eec476492d634752898f8285cb6e1222 Mon Sep 17 00:00:00 2001 From: Clemens Vasters Date: Fri, 13 Nov 2026 11:42:08 +0100 Subject: [PATCH] Honor inherited Avro media types in Python producers Preserve JSON and binary Avro variants over the same schema, normalize the reported Avro alias only for serialization, and retain datacontenttype in structured Kafka envelopes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: dfdd160a-bd4a-4a7e-bfb0-34d189fc944a Signed-off-by: Clemens Vasters --- test/py/test_python.py | 108 +++++++++++++++++- .../py/_common/cloudevents.jinja.include | 9 ++ ...tdir!dotunderscore!lower}producer.py.jinja | 1 + ...tdir!dotunderscore!lower}producer.py.jinja | 2 +- ...tdir!dotunderscore!lower}producer.py.jinja | 9 +- ...ectdir!dotunderscore!lower}client.py.jinja | 4 +- ...tdir!dotunderscore!lower}producer.py.jinja | 2 +- 7 files changed, 127 insertions(+), 8 deletions(-) diff --git a/test/py/test_python.py b/test/py/test_python.py index c81ad053..9578f4ef 100644 --- a/test/py/test_python.py +++ b/test/py/test_python.py @@ -1191,6 +1191,110 @@ def test_kafkaproducer_time_uritemplate_is_normalized_before_cloudevent_creation assert "from_binary(message, data_unmarshaller=lambda data: data)" in generated_test_src +@pytest.mark.parametrize("style", ["kafkaproducer", "amqpproducer"]) +def test_python_producer_inherited_avro_content_type(style): + """Inherited media types select the serializer, not the Avro schema format.""" + import base64 + import glob + import inspect + import io + import py_compile + import avro.io + from unittest.mock import Mock + + avro_type = "application/vnd.apache.org.avro+avro" + base = { + "messageid": "Example.Post", + "envelope": "CloudEvents/1.0", + "envelopemetadata": { + "type": {"value": "Example.Post"}, + "source": {"value": "urn:test"}, + "subject": {"value": "post"}, + "datacontenttype": {"value": "application/json"}, + }, + "dataschemaformat": "Avro/1.11.1", + "dataschemauri": "#/schemagroups/Example/schemas/Example.Post", + } + document = {"messagegroups": { + "Base": {"messages": {"Example.Post": base}}, + "Avro": {"messages": {"Example.Post": { + "basemessage": "/messagegroups/Base/messages/Example.Post", + "envelopemetadata": {"datacontenttype": {"value": avro_type}}, + }}}, + "Inherited": {"messages": {"Example.Post": { + "basemessage": "/messagegroups/Avro/messages/Example.Post", + }}}, + }} + document["schemagroups"] = {"Example": {"schemas": {"Example.Post": { + "format": "Avro/1.11.1", + "versions": {"1": { + "format": "Avro/1.11.1", + "schema": { + "type": "record", + "name": "Post", + "namespace": "Example", + "fields": [{"name": "text", "type": "string"}], + }, + }}, + }}}} + directory, producer_file = _generate_python_entrypoint_from_document( + document, style, "producer.py") + for path in glob.glob(os.path.join(directory, "**", "*.py"), recursive=True): + py_compile.compile(path, doraise=True) + if style == "amqpproducer": + import ast + with open(producer_file, encoding="utf-8") as handle: + tree = ast.parse(handle.read()) + defaults = {} + for node in tree.body: + if isinstance(node, ast.ClassDef): + for method in node.body: + if isinstance(method, ast.FunctionDef) and method.name == "send_post": + arguments = method.args.args[-len(method.args.defaults):] + defaults[node.name] = dict(zip( + [arg.arg for arg in arguments], method.args.defaults))["content_type"].value + assert defaults["BaseProducer"] == "application/json" + assert defaults["AvroProducer"] == avro_type + assert defaults["InheritedProducer"] == avro_type + return + + paths = glob.glob(os.path.join(directory, "**", "src"), recursive=True) + module = _load_generated_python_module_from_path(producer_file, extra_paths=paths) + data = module.Post(text="hello") + for group, default_type in [("Base", "application/json"), ("Avro", avro_type), + ("Inherited", avro_type)]: + producer_class = getattr(module, f"{group}EventProducer") + assert inspect.signature(producer_class.send_example_post).parameters[ + "content_type"].default == default_type + for mode in ("binary", "structured"): + for override in ( + None, "application/json", avro_type, "avro/binary", + "application/vnd.apache.avro+avro", avro_type + "; charset=binary", + ): + recording = Mock() + producer = producer_class(recording, "posts", content_mode=mode) + kwargs = {"content_type": override} if override else {} + producer.send_example_post(data=data, flush_producer=False, **kwargs) + sent = recording.produce.call_args.kwargs + content_type = override or default_type + if mode == "binary": + assert sent["headers"]["content-type"] == content_type.encode("utf-8") + payload = sent["value"] + else: + assert sent["headers"]["content-type"] == b"application/cloudevents+json" + event = json.loads(sent["value"]) + assert event["datacontenttype"] == content_type + payload = base64.b64decode(event["data_base64"]) + assert "data" not in event + if content_type == "application/json": + assert json.loads(payload)["text"] == "hello" + else: + stream = io.BytesIO(payload) + assert avro.io.DatumReader(data.AvroType).read( + avro.io.BinaryDecoder(stream)) == {"text": "hello"} + assert stream.tell() == len(payload) + + def test_kafkaproducer_exposes_resilient_flush_helper(): """Regression for #430: the generated Kafka producer must expose a ``flush(timeout, retries, backoff_factor)`` helper that retries with @@ -1574,7 +1678,7 @@ def test_mqttclient_body_is_bytes_not_str(): have to double-decode JSON. """ src = _generate_mqtt_client_src() - assert "byte_data = data.to_byte_array(content_type)" in src, ( + assert "byte_data = data.to_byte_array(content_type.replace(" in src, ( "client.py must call to_byte_array on the dataclass" ) # The coercion guard must be immediately downstream. @@ -1792,7 +1896,7 @@ def test_ehproducer_body_is_bytes_not_str(): data section, not a string section containing escaped JSON. """ src = _generate_eh_producer_src("test/xreg/lightbulb.xreg.json") - assert "byte_data = data.to_byte_array(content_type)" in src + assert "byte_data = data.to_byte_array(content_type.replace(" in src assert "isinstance(byte_data, str)" in src and "byte_data.encode('utf-8')" in src, ( "ehproducer must coerce to_byte_array str result to bytes before " "handing to the CloudEvent constructor" diff --git a/xrcg/templates/py/_common/cloudevents.jinja.include b/xrcg/templates/py/_common/cloudevents.jinja.include index 09cf081c..4a32f147 100644 --- a/xrcg/templates/py/_common/cloudevents.jinja.include +++ b/xrcg/templates/py/_common/cloudevents.jinja.include @@ -14,7 +14,11 @@ {%- macro DefaultContentType(message) -%} {%- set schema_format = message.dataschemaformat | default("", true) | lower -%} {%- if schema_format.startswith("avro") -%} +{%- if message.envelopemetadata is defined and message.envelopemetadata and "datacontenttype" in message.envelopemetadata and message.envelopemetadata.datacontenttype.value -%} +'{{ message.envelopemetadata.datacontenttype.value }}' +{%- else -%} 'avro/binary' +{%- endif -%} {%- elif schema_format.startswith("proto") -%} 'application/protobuf' {%- elif schema_format.startswith("xsd") or schema_format.startswith("xmlschema") -%} @@ -32,6 +36,11 @@ {%- endif -%} {%- endmacro -%} +{# Normalize the Avro media-type alias for Avrotize, not for wire metadata. #} +{%- macro SerializerContentType(content_type) -%} +{{ content_type }}.replace("application/vnd.apache.org.avro+avro", "application/vnd.apache.avro+avro", 1) +{%- endmacro -%} + {%- macro DeclareRfc3339TimeNormalizer() -%} _RFC3339_TIMESTAMP_PATTERN = re.compile( r'^\d{4}-\d{2}-\d{2}[Tt]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[Zz]|[+-]\d{2}:\d{2})?$' diff --git a/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index fdbc91f7..c948a3c0 100644 --- a/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/amqpproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -664,6 +664,7 @@ class {{ class_name }}: if data is None: return b'' if hasattr(data, 'to_byte_array'): + content_type = {{ cloudEvents.SerializerContentType("content_type") }} payload = data.to_byte_array(content_type) elif hasattr(data, 'to_dict'): payload = json.dumps(data.to_dict()) diff --git a/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index 345cad98..9c54005d 100644 --- a/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/ehproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -131,7 +131,7 @@ class {{ class_name }}: } attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' # to_byte_array returns str for text content types (e.g. JSON); # coerce to bytes so the cloudevents-sdk emits a binary AMQP data # section rather than a string section containing escaped JSON. diff --git a/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index a2570367..ea4500ed 100644 --- a/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/kafkaproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -229,7 +229,7 @@ class {{ class_name }}: attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) {% if data_type != "object" -%} - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' {%- else -%} byte_data = json.dumps(data).encode('utf-8') if data is not None else b'' {%- endif %} @@ -237,7 +237,12 @@ class {{ class_name }}: byte_data = byte_data.encode('utf-8') event = CloudEvent.create(attributes, byte_data) if self.content_mode == "structured": - message = to_structured(event, data_marshaller=lambda x: self.__binary_data_marshaller(x), key_mapper=lambda x: self.__key_mapper(x, data, key_mapper, kafka_key)) + message = to_structured( + event, + data_marshaller=lambda x: self.__binary_data_marshaller(x), + envelope_marshaller=lambda envelope: json.dumps({**envelope, "datacontenttype": content_type}), + key_mapper=lambda x: self.__key_mapper(x, data, key_mapper, kafka_key), + ) message.headers["content-type"] = b"application/cloudevents+json" else: # For binary mode, datacontenttype is already set in attributes above diff --git a/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja b/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja index 7c57e7c7..8bbbb7fb 100644 --- a/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja +++ b/xrcg/templates/py/mqttclient/src/{mainprojectdir!dotunderscore!lower}client.py.jinja @@ -823,7 +823,7 @@ class {{ class_name }}(_ClientBase): } attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' # to_byte_array returns str for text content types (e.g. JSON); # paho-mqtt will UTF-8 encode str payloads, but cloudevents-sdk's # to_binary/to_structured embed the str directly which then becomes @@ -953,7 +953,7 @@ class {{ class_name }}(_ClientBase): else: payload = json.dumps(data).encode('utf-8') {%- else %} - payload = data.to_byte_array(content_type) if data is not None else b'' + payload = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' if isinstance(payload, str): payload = payload.encode('utf-8') {%- endif %} diff --git a/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja b/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja index 3487f379..4769f8c9 100644 --- a/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja +++ b/xrcg/templates/py/sbproducer/src/{mainprojectdir!dotunderscore!lower}producer.py.jinja @@ -150,7 +150,7 @@ class {{ class_name }}: } attributes["datacontenttype"] = content_type attributes["time"] = _resolve_cloudevents_time(_time, attributes.get("time")) - byte_data = data.to_byte_array(content_type) if data is not None else b'' + byte_data = data.to_byte_array({{ cloudEvents.SerializerContentType("content_type") }}) if data is not None else b'' # to_byte_array returns str for text content types (e.g. JSON); coerce # to bytes so the wire body is the exact serialized representation # (binary AMQP data section) and receivers do not double-decode JSON.