fix Topic.deserialize not using the correct value deserializer (#413)

Author: tim-quix

quixstreams/models/topics/topic.py

@@ -274,19 +274,23 @@ def serialize(
 
     def deserialize(self, message: ConfluentKafkaMessageProto):
         ctx = SerializationContext(topic=message.topic(), headers=message.headers())
-        key_bytes = message.key()
-        value_bytes = message.value()
+        if (key := message.key()) is not None:
+            if self._key_deserializer:
+                key = self._key_deserializer(key, ctx=ctx)
+            else:
+                raise DeserializerIsNotProvidedError(
+                    f'Key deserializer is not provided for topic "{self.name}"'
+                )
+        if (value := message.value()) is not None:
+            if self._value_deserializer:
+                value = self._value_deserializer(value, ctx=ctx)
+            else:
+                raise DeserializerIsNotProvidedError(
+                    f'Value deserializer is not provided for topic "{self.name}"'
+                )
         return KafkaMessage(
-            key=(
-                None
-                if key_bytes is None
-                else self._key_deserializer(key_bytes, ctx=ctx)
-            ),
-            value=(
-                None
-                if value_bytes is None
-                else self._value_serializer(value_bytes, ctx=ctx)
-            ),
+            key=key,
+            value=value,
             headers=message.headers(),
             timestamp=message.timestamp()[1],
         )

tests/test_quixstreams/fixtures.py

@@ -580,6 +580,7 @@ def factory(
         name: Optional[str] = None,
         partitions: int = 1,
         create_topic: bool = False,
+        use_serdes_nones: bool = False,
         key_serializer: Optional[Union[Serializer, str]] = None,
         value_serializer: Optional[Union[Serializer, str]] = None,
         key_deserializer: Optional[Union[Deserializer, str]] = None,
@@ -596,9 +597,10 @@ def factory(
             "config": topic_manager.topic_config(num_partitions=partitions),
             "timestamp_extractor": timestamp_extractor,
         }
-        topic = topic_manager.topic(
-            name, **{k: v for k, v in topic_args.items() if v is not None}
-        )
+        if not use_serdes_nones:
+            # will use the topic manager serdes defaults rather than "Nones"
+            topic_args = {k: v for k, v in topic_args.items() if v is not None}
+        topic = topic_manager.topic(name, **topic_args)
         if create_topic:
             topic_manager.create_all_topics()
         return topic

tests/test_quixstreams/test_models/test_topics/test_topics.py

@@ -502,6 +502,55 @@ def test_serialize_serializer_missing(
                 timestamp_ms=1234567890,
             )
 
+    @pytest.mark.parametrize(
+        "deserializers, input_kv, expected_kv",
+        [
+            (
+                {"key_deserializer": "str", "value_deserializer": "json"},
+                {"key": b"woo", "value": b'{"a":["cool","json"]}'},
+                {"key": "woo", "value": {"a": ["cool", "json"]}},
+            ),
+            (
+                {"key_deserializer": None, "value_deserializer": "int"},
+                {"key": None, "value": int_to_bytes(12345)},
+                {"key": None, "value": 12345},
+            ),
+            (
+                {"key_deserializer": "str", "value_deserializer": "json"},
+                {"key": None, "value": None},
+                {"key": None, "value": None},
+            ),
+        ],
+    )
+    def test_deserialize(
+        self, topic_manager_topic_factory, deserializers, input_kv, expected_kv
+    ):
+        topic = topic_manager_topic_factory(**deserializers, use_serdes_nones=True)
+        message = topic.deserialize(
+            ConfluentKafkaMessageStub(
+                **input_kv,
+                headers=[("header", b"value")],
+                timestamp=(1000, 2000),
+            )
+        )
+        assert {"key": message.key, "value": message.value} == expected_kv
+        assert message.headers == [("header", b"value")]
+        assert message.timestamp == 2000
+
+    def test_deserializer_missing_deserializer(self, topic_manager_topic_factory):
+        topic = topic_manager_topic_factory(
+            key_deserializer=None, value_deserializer="str", use_serdes_nones=True
+        )
+        with pytest.raises(DeserializerIsNotProvidedError):
+            topic.deserialize(
+                ConfluentKafkaMessageStub(
+                    key=b"key",
+                    value=b"value",
+                    headers=[("header", b"value")],
+                    timestamp=(1000, 2000),
+                )
+            )
+
     @pytest.mark.parametrize(
         "key_serializer, value_serializer, key, value",
         [