feat(python): add topic listing, update, delete and purge

foreign/python/apache_iggy.pyi

@@ -34,6 +34,7 @@ __all__ = [
     "ReceiveMessage",
     "SendMessage",
     "StreamDetails",
+    "Topic",
     "TopicDetails",
 ]
 
@@ -302,6 +303,88 @@ class IggyClient:
         Gets topic by stream and id.
         Returns Option of topic details or a PyRuntimeError on failure.
         """
+    def get_topics(
+        self, stream_id: builtins.str | builtins.int
+    ) -> collections.abc.Awaitable[list[Topic]]:
+        r"""
+        Get all topics in a stream.
+
+        Args:
+            stream_id: Stream identifier as `str | int`.
+
+        Returns:
+            An awaitable that resolves to `list[Topic]`.
+
+        Raises:
+            PyRuntimeError: If the identifier is invalid or the request fails.
+        """
+    def update_topic(
+        self,
+        stream_id: builtins.str | builtins.int,
+        topic_id: builtins.str | builtins.int,
+        name: builtins.str,
+        compression_algorithm: builtins.str | None = None,
+        replication_factor: builtins.int | None = None,
+        message_expiry: datetime.timedelta | None = None,
+        max_topic_size: builtins.int | None = None,
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Update an existing topic.
+
+        This is a full replacement: any optional parameter left unset is reset to
+        its server default rather than preserved.
+
+        Args:
+            stream_id: Stream identifier as `str | int`.
+            topic_id: Topic identifier as `str | int`.
+            name: New topic name as `str`.
+            compression_algorithm: Compression algorithm as `str | None`.
+            replication_factor: Replication factor as `int | None`.
+            message_expiry: Message expiry as `datetime.timedelta | None`.
+            max_topic_size: Maximum topic size in bytes as `int | None`.
+
+        Returns:
+            An awaitable that resolves to `None` when the topic is updated.
+
+        Raises:
+            PyRuntimeError: If an argument is invalid or the request fails.
+        """
+    def delete_topic(
+        self,
+        stream_id: builtins.str | builtins.int,
+        topic_id: builtins.str | builtins.int,
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Delete a topic from a stream.
+
+        Args:
+            stream_id: Stream identifier as `str | int`.
+            topic_id: Topic identifier as `str | int`.
+
+        Returns:
+            An awaitable that resolves to `None` when the topic is deleted.
+
+        Raises:
+            PyRuntimeError: If an identifier is invalid or the request fails.
+        """
+    def purge_topic(
+        self,
+        stream_id: builtins.str | builtins.int,
+        topic_id: builtins.str | builtins.int,
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Purge all messages from a topic.
+
+        Args:
+            stream_id: Stream identifier as `str | int`.
+            topic_id: Topic identifier as `str | int`.
+
+        Returns:
+            An awaitable that resolves to `None` when the topic is purged.
+
+        Raises:
+            PyRuntimeError: If an identifier is invalid or the request fails.
+        """
     def send_messages(
         self,
         stream: builtins.str | builtins.int,
@@ -519,6 +602,17 @@ class StreamDetails:
     @property
     def topics_count(self) -> builtins.int: ...
 
+@typing.final
+class Topic:
+    @property
+    def id(self) -> builtins.int: ...
+    @property
+    def name(self) -> builtins.str: ...
+    @property
+    def messages_count(self) -> builtins.int: ...
+    @property
+    def partitions_count(self) -> builtins.int: ...
+
 @typing.final
 class TopicDetails:
     @property
@@ -529,3 +623,7 @@ class TopicDetails:
     def messages_count(self) -> builtins.int: ...
     @property
     def partitions_count(self) -> builtins.int: ...
+    @property
+    def compression_algorithm(self) -> builtins.str: ...
+    @property
+    def replication_factor(self) -> builtins.int: ...

foreign/python/src/client.rs

@@ -33,7 +33,7 @@ use crate::identifier::PyIdentifier;
 use crate::receive_message::{PollingStrategy, ReceiveMessage};
 use crate::send_message::SendMessage;
 use crate::stream::StreamDetails;
-use crate::topic::TopicDetails;
+use crate::topic::{Topic, TopicDetails};
 use tokio::sync::Mutex;
 
 /// A Python class representing the Iggy client.
@@ -243,6 +243,170 @@ impl IggyClient {
         })
     }
 
+    /// Get all topics in a stream.
+    ///
+    /// Args:
+    ///     stream_id: Stream identifier as `str | int`.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `list[Topic]`.
+    ///
+    /// Raises:
+    ///     PyRuntimeError: If the identifier is invalid or the request fails.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[Topic]]", imports=("collections.abc")))]
+    fn get_topics<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let stream_id = Identifier::try_from(stream_id)?;
+        let inner = self.inner.clone();
+
+        future_into_py(py, async move {
+            let topics = inner
+                .get_topics(&stream_id)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(topics.into_iter().map(Topic::from).collect::<Vec<_>>())
+        })
+    }
+
+    /// Update an existing topic.
+    ///
+    /// This is a full replacement: any optional parameter left unset is reset to
+    /// its server default rather than preserved.
+    ///
+    /// Args:
+    ///     stream_id: Stream identifier as `str | int`.
+    ///     topic_id: Topic identifier as `str | int`.
+    ///     name: New topic name as `str`.
+    ///     compression_algorithm: Compression algorithm as `str | None`.
+    ///     replication_factor: Replication factor as `int | None`.
+    ///     message_expiry: Message expiry as `datetime.timedelta | None`.
+    ///     max_topic_size: Maximum topic size in bytes as `int | None`.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `None` when the topic is updated.
+    ///
+    /// Raises:
+    ///     PyRuntimeError: If an argument is invalid or the request fails.
+    #[pyo3(
+        signature = (stream_id, topic_id, name, compression_algorithm = None, replication_factor = None, message_expiry = None, max_topic_size = None)
+    )]
+    #[allow(clippy::too_many_arguments)]
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn update_topic<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+        topic_id: PyIdentifier,
+        name: String,
+        #[gen_stub(override_type(type_repr = "builtins.str | None"))] compression_algorithm: Option<
+            String,
+        >,
+        #[gen_stub(override_type(type_repr = "builtins.int | None"))] replication_factor: Option<
+            u8,
+        >,
+        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+        message_expiry: Option<Py<PyDelta>>,
+        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_topic_size: Option<u64>,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let compression_algorithm = match compression_algorithm {
+            Some(algo) => CompressionAlgorithm::from_str(&algo)
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
+            None => CompressionAlgorithm::default(),
+        };
+
+        let expiry = match message_expiry {
+            Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)),
+            None => IggyExpiry::ServerDefault,
+        };
+
+        let max_size = max_topic_size.map_or(MaxTopicSize::ServerDefault, MaxTopicSize::from);
+
+        let stream_id = Identifier::try_from(stream_id)?;
+        let topic_id = Identifier::try_from(topic_id)?;
+        let inner = self.inner.clone();
+
+        future_into_py(py, async move {
+            inner
+                .update_topic(
+                    &stream_id,
+                    &topic_id,
+                    &name,
+                    compression_algorithm,
+                    replication_factor,
+                    expiry,
+                    max_size,
+                )
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
+    /// Delete a topic from a stream.
+    ///
+    /// Args:
+    ///     stream_id: Stream identifier as `str | int`.
+    ///     topic_id: Topic identifier as `str | int`.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `None` when the topic is deleted.
+    ///
+    /// Raises:
+    ///     PyRuntimeError: If an identifier is invalid or the request fails.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn delete_topic<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+        topic_id: PyIdentifier,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let stream_id = Identifier::try_from(stream_id)?;
+        let topic_id = Identifier::try_from(topic_id)?;
+        let inner = self.inner.clone();
+
+        future_into_py(py, async move {
+            inner
+                .delete_topic(&stream_id, &topic_id)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
+    /// Purge all messages from a topic.
+    ///
+    /// Args:
+    ///     stream_id: Stream identifier as `str | int`.
+    ///     topic_id: Topic identifier as `str | int`.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `None` when the topic is purged.
+    ///
+    /// Raises:
+    ///     PyRuntimeError: If an identifier is invalid or the request fails.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn purge_topic<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+        topic_id: PyIdentifier,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let stream_id = Identifier::try_from(stream_id)?;
+        let topic_id = Identifier::try_from(topic_id)?;
+        let inner = self.inner.clone();
+
+        future_into_py(py, async move {
+            inner
+                .purge_topic(&stream_id, &topic_id)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
     /// Sends a list of messages to the specified topic.
     /// Returns Ok(()) on successful sending or a PyRuntimeError on failure.
     #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]

foreign/python/src/lib.rs

@@ -29,7 +29,7 @@ use pyo3::prelude::*;
 use receive_message::{PollingStrategy, ReceiveMessage};
 use send_message::SendMessage;
 use stream::StreamDetails;
-use topic::TopicDetails;
+use topic::{Topic, TopicDetails};
 
 /// A Python module implemented in Rust.
 #[pymodule]
@@ -38,6 +38,7 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<ReceiveMessage>()?;
     m.add_class::<IggyClient>()?;
     m.add_class::<StreamDetails>()?;
+    m.add_class::<Topic>()?;
     m.add_class::<TopicDetails>()?;
     m.add_class::<PollingStrategy>()?;
     m.add_class::<IggyConsumer>()?;

foreign/python/src/topic.rs

@@ -15,10 +15,46 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use iggy::prelude::TopicDetails as RustTopicDetails;
+use iggy::prelude::{Topic as RustTopic, TopicDetails as RustTopicDetails};
 use pyo3::prelude::*;
 use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
 
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct Topic {
+    pub(crate) inner: RustTopic,
+}
+
+impl From<RustTopic> for Topic {
+    fn from(topic: RustTopic) -> Self {
+        Self { inner: topic }
+    }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Topic {
+    #[getter]
+    pub fn id(&self) -> u32 {
+        self.inner.id
+    }
+
+    #[getter]
+    pub fn name(&self) -> String {
+        self.inner.name.to_string()
+    }
+
+    #[getter]
+    pub fn messages_count(&self) -> u64 {
+        self.inner.messages_count
+    }
+
+    #[getter]
+    pub fn partitions_count(&self) -> u32 {
+        self.inner.partitions_count
+    }
+}
+
 #[gen_stub_pyclass]
 #[pyclass]
 pub struct TopicDetails {
@@ -55,4 +91,14 @@ impl TopicDetails {
     pub fn partitions_count(&self) -> u32 {
         self.inner.partitions_count
     }
+
+    #[getter]
+    pub fn compression_algorithm(&self) -> String {
+        self.inner.compression_algorithm.to_string()
+    }
+
+    #[getter]
+    pub fn replication_factor(&self) -> u8 {
+        self.inner.replication_factor
+    }
 }