Update Topic name handling with Quix platform

- Added new field "Topic.quix_name" to store the non-prefixed topic name from Quix
- Generate stream_ids in backwards-compatible way for

Author: daniil-quix

quixstreams/dataframe/dataframe.py

@@ -172,9 +172,7 @@ def stream_id(self) -> str:
 
         By default, a topic name or a combination of topic names are used as `stream_id`.
         """
-
-        topics = sorted(self._topics, key=lambda t: t.name)
-        return "--".join(t.name for t in topics)
+        return self._topic_manager.stream_id_from_topics(*self.topics)
 
     @property
     def topics(self) -> list[Topic]:

quixstreams/models/topics/manager.py

@@ -332,6 +332,15 @@ def derive_topic_config(cls, *topics: Topic) -> TopicConfig:
             },
         )
 
+    def stream_id_from_topics(self, *topics: Topic) -> str:
+        """
+        Generate a stream_id by combining names of the provided topics.
+        """
+        if not topics:
+            raise ValueError("At least one Topic must be passed")
+
+        return "--".join(t.name for t in topics)
+
     def _validate_topic_name(self, name: str) -> None:
         """
         Validates the original topic name

quixstreams/models/topics/topic.py

@@ -100,6 +100,7 @@ def __init__(
         value_serializer: Optional[SerializerType] = None,
         key_serializer: Optional[SerializerType] = BytesSerializer(),
         timestamp_extractor: Optional[TimestampExtractor] = None,
+        quix_name: str = "",
     ):
         """
         :param name: topic name
@@ -110,8 +111,11 @@ def __init__(
         :param key_serializer: a serializer type for keys
         :param timestamp_extractor: a callable that returns a timestamp in
             milliseconds from a deserialized message.
+        :param quix_name: a name of the topic in the Quix Cloud.
+            It is set only by `QuixTopicManager`.
         """
         self.name = name
+        self.quix_name = quix_name or name
         self._create_config = copy.deepcopy(create_config)
         self._broker_config: Optional[TopicConfig] = None
         self._value_deserializer = _get_deserializer(value_deserializer)
@@ -127,6 +131,7 @@ def __clone__(
     ):
         return self.__class__(
             name=name,
+            quix_name=self.quix_name,
             create_config=create_config or self._create_config,
             value_deserializer=self._value_deserializer,
             key_deserializer=self._key_deserializer,

quixstreams/platforms/quix/config.py

@@ -187,8 +187,6 @@ def convert_topic_response(cls, api_response: dict) -> Topic:
         topic_config = api_response["configuration"]
         extra_config["retention.ms"] = topic_config["retentionInMinutes"] * 60 * 1000
         extra_config["retention.bytes"] = topic_config["retentionInBytes"]
-        # A hack to pass extra info back from Quix cloud
-        extra_config["__quix_topic_name__"] = api_response["name"]
 
         # Map value returned by Quix API to Kafka Admin API format
         if topic_config.get("cleanupPolicy"):
@@ -202,7 +200,11 @@ def convert_topic_response(cls, api_response: dict) -> Topic:
             replication_factor=topic_config["replicationFactor"],
             extra_config=extra_config,
         )
-        topic = Topic(name=api_response["id"], create_config=config)
+        topic = Topic(
+            name=api_response["id"],
+            create_config=config,
+            quix_name=api_response["name"],
+        )
         topic.broker_config = config
         return topic
 

quixstreams/platforms/quix/topic_manager.py

@@ -59,7 +59,21 @@ def __init__(
             auto_create_topics=auto_create_topics,
         )
         self._quix_config_builder = quix_config_builder
-        self._topic_id_to_name: dict[str, str] = {}
+
+    def stream_id_from_topics(self, *topics: Topic) -> str:
+        """
+        Generate a stream_id by combining names of the provided topics.
+        """
+        if not topics:
+            raise ValueError("At least one Topic must be passed")
+        elif len(topics) == 1:
+            # If only one topic is passed, return its full name
+            # for backwards compatibility
+            return topics[0].name
+
+        # Use the "quix_name" to generate stream_id.
+        # In Quix Cloud, the "quix_name" can differ from the actual broker topic name
+        return "--".join(t.quix_name for t in topics)
 
     def _fetch_topic(self, topic: Topic) -> Topic:
         try:
@@ -92,10 +106,6 @@ def _finalize_topic(self, topic: Topic) -> Topic:
         broker_topic = self._fetch_topic(topic=topic)
         broker_config = broker_topic.broker_config
 
-        # A hack to pass extra info back from Quix cloud
-        quix_topic_name = broker_config.extra_config.pop("__quix_topic_name__")
-        topic_out = topic.__clone__(name=broker_topic.name)
-
         # Set a broker config for the topic
         broker_config = TopicConfig(
             num_partitions=broker_config.num_partitions,
@@ -106,8 +116,8 @@ def _finalize_topic(self, topic: Topic) -> Topic:
                 if k in self._extra_config_imports
             },
         )
+        topic_out = topic.__clone__(name=broker_topic.name)
         topic_out.broker_config = broker_config
-        self._topic_id_to_name[topic_out.name] = quix_topic_name
         self._quix_config_builder.wait_for_topic_ready_statuses([topic_out])
         return topic_out
 
@@ -148,8 +158,13 @@ def _internal_name(
 
         :return: formatted topic name
         """
-        return super()._internal_name(
-            topic_type,
-            self._topic_id_to_name[topic_name] if topic_name else None,
-            suffix,
-        )
+
+        # Map the full topic name to the shorter "quix_name" and use
+        # it for internal topics.
+        # "quix_name" is not prefixed with the workspace id.
+        if topic_name is not None:
+            topic = self.non_changelog_topics.get(topic_name)
+            if topic is not None:
+                topic_name = topic.quix_name
+
+        return super()._internal_name(topic_type, topic_name, suffix)

tests/test_quixstreams/fixtures.py

@@ -600,9 +600,10 @@ def factory(
         key_deserializer: Optional[Union[Deserializer, str]] = None,
         value_deserializer: Optional[Union[Deserializer, str]] = None,
         timestamp_extractor: Optional[TimestampExtractor] = None,
+        topic_manager: Optional[TopicManager] = None,
     ) -> Topic:
         name = name or str(uuid.uuid4())
-        topic_manager = topic_manager_factory()
+        topic_manager = topic_manager or topic_manager_factory()
         topic_args = {
             "key_serializer": key_serializer,
             "value_serializer": value_serializer,

tests/test_quixstreams/test_dataframe/fixtures.py

@@ -26,9 +26,11 @@ def factory(
         registry: Optional[DataFrameRegistry] = None,
     ) -> StreamingDataFrame:
         producer = producer if producer is not None else MagicMock(spec_set=RowProducer)
-        topic_manager = topic_manager or MagicMock(spec=TopicManager)
+        topic_manager = topic_manager or topic_manager_factory()
         state_manager = state_manager or MagicMock(spec=StateStoreManager)
-        topic = topic or topic_manager_topic_factory("test")
+        topic = topic or topic_manager_topic_factory(
+            "test", topic_manager=topic_manager
+        )
         consumer = MagicMock(spec_set=RowConsumer)
         pausing_manager = PausingManager(consumer=consumer, topic_manager=topic_manager)
         sink_manager = SinkManager()

tests/test_quixstreams/test_models/test_topics/test_manager.py

@@ -49,6 +49,7 @@ def test_topic_with_config(self, topic_manager_factory):
         assert topic_manager.topics[topic_name] == topic
 
         assert topic.name == topic_name
+        assert topic.quix_name == topic_name
         assert topic.create_config.num_partitions == create_config.num_partitions
         assert (
             topic.create_config.replication_factor == create_config.replication_factor
@@ -321,3 +322,16 @@ def test_non_changelog_topics(self, topic_manager_factory):
         assert data_topic.name in topic_manager.non_changelog_topics
         assert repartition_topic.name in topic_manager.non_changelog_topics
         assert changelog_topic.name not in topic_manager.non_changelog_topics
+
+    def test_stream_id_from_topics_success(self, topic_manager_factory):
+        topic_manager = topic_manager_factory()
+        topic1 = topic_manager.topic("test1")
+        topic2 = topic_manager.topic("test2")
+        stream_id = topic_manager.stream_id_from_topics(topic1, topic2)
+
+        assert stream_id == "test1--test2"
+
+    def test_stream_id_from_topics_no_topics_fails(self, topic_manager_factory):
+        topic_manager = topic_manager_factory()
+        with pytest.raises(ValueError):
+            topic_manager.stream_id_from_topics()

tests/test_quixstreams/test_platforms/test_quix/test_topic_manager.py

@@ -1,3 +1,5 @@
+import pytest
+
 from quixstreams.models import TopicConfig
 
 
@@ -13,6 +15,7 @@ def test_quix_topic(self, quix_topic_manager_factory):
         topic_manager = quix_topic_manager_factory()
         topic = topic_manager.topic(topic_name, create_config=create_config)
         assert topic.name == expected_topic_id
+        assert topic.quix_name == topic_name
         assert topic_manager.topics[topic.name] == topic
         assert topic.broker_config.num_partitions == create_config.num_partitions
         assert (
@@ -115,3 +118,31 @@ def test_quix_changelog_nested_internal_topic_naming(
             changelog.broker_config.replication_factor
             == repartition.broker_config.replication_factor
         )
+
+    def test_stream_id_from_topics_multiple_topics_success(
+        self, quix_topic_manager_factory
+    ):
+        topic_manager = quix_topic_manager_factory(workspace_id="workspace_id")
+        topic1 = topic_manager.topic("test1")
+        topic2 = topic_manager.topic("test2")
+        stream_id = topic_manager.stream_id_from_topics(topic1, topic2)
+
+        assert stream_id == "test1--test2"
+
+    def test_stream_id_from_topics_single_topic_prefixed_with_workspace(
+        self, quix_topic_manager_factory
+    ):
+        """
+        Test that stream_id is prefixed with workspace_id if the single topic is passed
+        for the backwards compatibility.
+        """
+        topic_manager = quix_topic_manager_factory(workspace_id="workspace_id")
+        topic1 = topic_manager.topic("test1")
+        stream_id = topic_manager.stream_id_from_topics(topic1)
+
+        assert stream_id == "workspace_id-test1"
+
+    def test_stream_id_from_topics_no_topics_fails(self, quix_topic_manager_factory):
+        topic_manager = quix_topic_manager_factory()
+        with pytest.raises(ValueError):
+            topic_manager.stream_id_from_topics()