Added some stream methods.

Author: s3rius

python/natsrpy/_inner/js/__init__.pyi

@@ -20,4 +20,4 @@ class JetStream:
     async def create_stream(self, config: StreamConfig) -> Stream: ...
     async def update_stream(self, config: StreamConfig) -> Stream: ...
     async def get_stream(self, name: str) -> Stream: ...
-    async def delete_stream(self, name: str) -> Stream: ...
+    async def delete_stream(self, name: str) -> bool: ...

python/natsrpy/_inner/js/stream.pyi

@@ -181,6 +181,50 @@ class StreamMessage:
     payload: bytes
     time: datetime
 
+class StreamState:
+    messages: int
+    bytes: int
+    first_sequence: int
+    first_timestamp: int
+    last_sequence: int
+    last_timestamp: int
+    consumer_count: int
+    subjects_count: int
+    deleted_count: int | None
+    deleted: list[int] | None
+
+class SourceInfo:
+    name: str
+    lag: int
+    active: timedelta | None
+    filter_subject: str | None
+    subject_transform_dest: str | None
+    subject_transforms: list[SubjectTransform]
+
+class PeerInfo:
+    name: str
+    current: bool
+    active: timedelta
+    offline: bool
+    lag: int | None
+
+class ClusterInfo:
+    name: str | None
+    raft_group: str | None
+    leader: str | None
+    leader_since: int | None
+    system_account: bool
+    traffic_account: str | None
+    replicas: list[PeerInfo]
+
+class StreamInfo:
+    config: StreamConfig
+    created: float
+    state: StreamState
+    cluster: ClusterInfo | None
+    mirror: SourceInfo | None
+    sources: list[SourceInfo]
+
 class Stream:
     async def direct_get(self, sequence: int) -> StreamMessage:
         """

python/natsrpy/js/stream.py

@@ -13,6 +13,11 @@
     StreamConfig,
     StreamMessage,
     SubjectTransform,
+    ClusterInfo,
+    PeerInfo,
+    SourceInfo,
+    StreamInfo,
+    StreamState,
 )
 
 __all__ = [
@@ -30,4 +35,16 @@
     "StreamConfig",
     "StreamMessage",
     "SubjectTransform",
+    "ClusterInfo",
+    "Compression",
+    "ConsumerLimits",
+    "DiscardPolicy",
+    "Stream",
+    "PeerInfo",
+    "PersistenceMode",
+    "RetentionPolicy",
+    "SourceInfo",
+    "StreamConfig",
+    "StreamInfo",
+    "StreamState",
 ]

src/exceptions/rust_err.rs

@@ -54,6 +54,8 @@ pub enum NatsrpyError {
     GetStreamError(#[from] async_nats::jetstream::context::GetStreamError),
     #[error(transparent)]
     StreamDirectGetError(#[from] async_nats::jetstream::stream::DirectGetError),
+    #[error(transparent)]
+    StreamInfoError(#[from] async_nats::jetstream::stream::InfoError),
 }
 
 impl From<NatsrpyError> for pyo3::PyErr {

src/js/consumers/mod.rs

@@ -0,0 +1,2 @@
+pub mod pull;
+pub mod push;

src/js/consumers/pull.rs

@@ -0,0 +1,3 @@
+#[pyo3::pyclass(from_py_object)]
+#[derive(Debug, Clone)]
+pub struct PullConsumer;

src/js/consumers/push.rs

@@ -0,0 +1,3 @@
+#[pyo3::pyclass(from_py_object)]
+#[derive(Debug, Clone)]
+pub struct PushConsumer;

src/js/jetstream.rs

@@ -2,7 +2,7 @@ use std::{ops::Deref, sync::Arc};
 
 use async_nats::{Subject, client::traits::Publisher, connection::State};
 use pyo3::{
-    Bound, PyAny, Python, pyclass, pymethods,
+    Bound, PyAny, Python,
     types::{PyBytes, PyBytesMethods, PyDict},
 };
 use tokio::sync::RwLock;
@@ -16,7 +16,7 @@ use crate::{
     utils::{headers::NatsrpyHeadermapExt, natsrpy_future},
 };
 
-#[pyclass]
+#[pyo3::pyclass]
 pub struct JetStream {
     ctx: Arc<RwLock<async_nats::jetstream::Context>>,
 }
@@ -30,7 +30,7 @@ impl JetStream {
     }
 }
 
-#[pymethods]
+#[pyo3::pymethods]
 impl JetStream {
     #[pyo3(signature = (
         subject,
@@ -120,6 +120,18 @@ impl JetStream {
         })
     }
 
+    pub fn get_stream<'py>(
+        &self,
+        py: Python<'py>,
+        name: String,
+    ) -> NatsrpyResult<Bound<'py, PyAny>> {
+        let ctx = self.ctx.clone();
+        natsrpy_future(py, async move {
+            let js = ctx.read().await;
+            Ok(super::stream::Stream::new(js.get_stream(name).await?))
+        })
+    }
+
     pub fn create_stream<'py>(
         &self,
         py: Python<'py>,
@@ -135,15 +147,15 @@ impl JetStream {
         })
     }
 
-    pub fn get_stream<'py>(
+    pub fn delete_stream<'py>(
         &self,
         py: Python<'py>,
         name: String,
     ) -> NatsrpyResult<Bound<'py, PyAny>> {
         let ctx = self.ctx.clone();
         natsrpy_future(py, async move {
             let js = ctx.read().await;
-            Ok(super::stream::Stream::new(js.get_stream(name).await?))
+            Ok(js.delete_stream(name).await?.success)
         })
     }
 }

src/js/kv.rs

@@ -2,7 +2,7 @@ use std::{sync::Arc, time::Duration};
 
 use crate::js;
 use pyo3::{
-    Bound, PyAny, Python, pyclass, pymethods,
+    Bound, PyAny, Python,
     types::{PyBytes, PyBytesMethods},
 };
 use tokio::sync::RwLock;
@@ -12,7 +12,7 @@ use crate::{
     utils::natsrpy_future,
 };
 
-#[pyclass(from_py_object, get_all, set_all)]
+#[pyo3::pyclass(from_py_object, get_all, set_all)]
 #[derive(Clone)]
 pub struct KVConfig {
     bucket: String,
@@ -32,7 +32,7 @@ pub struct KVConfig {
     limit_markers: Option<Duration>,
 }
 
-#[pymethods]
+#[pyo3::pymethods]
 impl KVConfig {
     #[new]
     #[pyo3(signature=(
@@ -128,7 +128,7 @@ impl TryFrom<KVConfig> for async_nats::jetstream::kv::Config {
     }
 }
 
-#[pyclass(from_py_object)]
+#[pyo3::pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct KeyValue {
     #[pyo3(get)]
@@ -159,7 +159,7 @@ impl KeyValue {
     }
 }
 
-#[pymethods]
+#[pyo3::pymethods]
 impl KeyValue {
     pub fn get<'py>(&self, py: Python<'py>, key: String) -> NatsrpyResult<Bound<'py, PyAny>> {
         let store = self.store.clone();

src/js/mod.rs

@@ -1,17 +1,19 @@
+pub mod consumers;
 pub mod jetstream;
 pub mod kv;
 pub mod stream;
 
 #[pyo3::pymodule(submodule, name = "js")]
 pub mod pymod {
+    // Classes
     #[pymodule_export]
-    pub use super::jetstream::JetStream;
-    #[pymodule_export]
-    pub use super::kv::KVConfig;
-
-    #[pymodule_export]
-    pub use super::kv::KeyValue;
+    pub use super::{
+        consumers::{pull::PullConsumer, push::PushConsumer},
+        jetstream::JetStream,
+        kv::{KVConfig, KeyValue},
+    };
 
+    // SubModules
     #[pymodule_export]
     pub use super::kv::pymod as kv;
     #[pymodule_export]