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

foreign/python/apache_iggy.pyi

@@ -36,6 +36,7 @@ __all__ = [
     "PollingStrategy",
     "ReceiveMessage",
     "SendMessage",
+    "Stream",
     "StreamDetails",
     "Topic",
     "TopicDetails",
@@ -352,6 +353,32 @@ class IggyClient:
         Gets stream by id.
         Returns Option of stream details or a PyRuntimeError on failure.
         """
+    def get_streams(self) -> collections.abc.Awaitable[list[Stream]]:
+        r"""
+        Gets all streams.
+        Returns a list of streams or a PyRuntimeError on failure.
+        """
+    def update_stream(
+        self, stream_id: builtins.str | builtins.int, name: builtins.str
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Updates a stream's name by id.
+        Returns Ok(()) on successful update or a PyRuntimeError on failure.
+        """
+    def delete_stream(
+        self, stream_id: builtins.str | builtins.int
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Deletes a stream by id.
+        Returns Ok(()) on successful deletion or a PyRuntimeError on failure.
+        """
+    def purge_stream(
+        self, stream_id: builtins.str | builtins.int
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Purges all messages from a stream by id.
+        Returns Ok(()) on successful purge or a PyRuntimeError on failure.
+        """
     def create_topic(
         self,
         stream: builtins.str | builtins.int,
@@ -796,6 +823,17 @@ class SendMessage:
         directly from Python using the provided string or bytes data.
         """
 
+@typing.final
+class Stream:
+    @property
+    def id(self) -> builtins.int: ...
+    @property
+    def name(self) -> builtins.str: ...
+    @property
+    def messages_count(self) -> builtins.int: ...
+    @property
+    def topics_count(self) -> builtins.int: ...
+
 @typing.final
 class StreamDetails:
     @property

foreign/python/src/client.rs

@@ -35,7 +35,7 @@ use crate::consumer::{
 use crate::identifier::PyIdentifier;
 use crate::receive_message::{PollingStrategy, ReceiveMessage};
 use crate::send_message::SendMessage;
-use crate::stream::StreamDetails;
+use crate::stream::{Stream, StreamDetails};
 use crate::topic::{Topic, TopicDetails};
 use tokio::sync::Mutex;
 
@@ -168,6 +168,78 @@ impl IggyClient {
         })
     }
 
+    /// Gets all streams.
+    /// Returns a list of streams or a PyRuntimeError on failure.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[Stream]]", imports=("collections.abc")))]
+    fn get_streams<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
+        let inner = self.inner.clone();
+        future_into_py(py, async move {
+            let streams = inner
+                .get_streams()
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(streams.into_iter().map(Stream::from).collect::<Vec<_>>())
+        })
+    }
+
+    /// Updates a stream's name by id.
+    /// Returns Ok(()) on successful update or a PyRuntimeError on failure.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn update_stream<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+        name: String,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let stream_id = Identifier::try_from(stream_id)?;
+        let inner = self.inner.clone();
+        future_into_py(py, async move {
+            inner
+                .update_stream(&stream_id, &name)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
+    /// Deletes a stream by id.
+    /// Returns Ok(()) on successful deletion or a PyRuntimeError on failure.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn delete_stream<'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 {
+            inner
+                .delete_stream(&stream_id)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
+    /// Purges all messages from a stream by id.
+    /// Returns Ok(()) on successful purge or a PyRuntimeError on failure.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn purge_stream<'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 {
+            inner
+                .purge_stream(&stream_id)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
     /// Creates a new topic with the given parameters.
     /// Returns Ok(()) on successful topic creation or a PyRuntimeError on failure.
     #[pyo3(

foreign/python/src/lib.rs

@@ -31,7 +31,7 @@ use consumer::{
 use pyo3::prelude::*;
 use receive_message::{PollingStrategy, ReceiveMessage};
 use send_message::SendMessage;
-use stream::StreamDetails;
+use stream::{Stream, StreamDetails};
 use topic::{Topic, TopicDetails};
 
 /// A Python module implemented in Rust.
@@ -41,6 +41,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::<Stream>()?;
     m.add_class::<Topic>()?;
     m.add_class::<TopicDetails>()?;
     m.add_class::<ConsumerGroup>()?;

foreign/python/src/stream.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use iggy::prelude::StreamDetails as RustStreamDetails;
+use iggy::prelude::{Stream as RustStream, StreamDetails as RustStreamDetails};
 use pyo3::prelude::*;
 use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
 
@@ -56,3 +56,39 @@ impl StreamDetails {
         self.inner.topics_count
     }
 }
+
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct Stream {
+    pub(crate) inner: RustStream,
+}
+
+impl From<RustStream> for Stream {
+    fn from(stream: RustStream) -> Self {
+        Self { inner: stream }
+    }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Stream {
+    #[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 topics_count(&self) -> u32 {
+        self.inner.topics_count
+    }
+}