Added stream config and stuff.

Author: s3rius

python/natsrpy/_inner/js/stream.pyi

@@ -1,22 +1,61 @@
+from datetime import timedelta
+
 class StorageType:
     FILE: "StorageType"
     MEMORY: "StorageType"
 
+class DiscardPolicy:
+    OLD: "DiscardPolicy"
+    NEW: "DiscardPolicy"
+
+class RetentionPolicy:
+    LIMITS: "RetentionPolicy"
+    INTEREST: "RetentionPolicy"
+    WORKQUEUE: "RetentionPolicy"
+
+class Compression:
+    S2: "Compression"
+    NONE: "Compression"
+
+class PersistenceMode:
+    Default: "PersistenceMode"
+    Async: "PersistenceMode"
+
+class ConsumerLimits:
+    inactive_threshold: timedelta
+    max_ack_pending: int
+
+    def __init__(self, inactive_threshold: timedelta, max_ack_pending: int) -> None: ...
+
 class External:
+    api_prefix: str
+    delivery_prefix: str | None
+
     def __init__(
         self,
         api_prefix: str,
         delivery_prefix: str | None = None,
     ) -> None: ...
 
 class SubjectTransform:
+    source: str
+    destination: str
+
     def __init__(
         self,
         source: str,
         destination: str,
     ) -> None: ...
 
 class Source:
+    name: str
+    filter_subject: str | None = None
+    external: External | None = None
+    start_sequence: int | None = None
+    start_time: int | None = None
+    domain: str | None = None
+    subject_transforms: SubjectTransform | None = None
+
     def __init__(
         self,
         name: str,
@@ -29,17 +68,67 @@ class Source:
     ) -> None: ...
 
 class Placement:
+    cluster: str | None
+    tags: list[str] | None
+
     def __init__(
         self,
         cluster: str | None = None,
         tags: list[str] | None = None,
     ) -> None: ...
 
 class Republish:
-    pass
+    source: str
+    destination: str
+    headers_only: bool
+
     def __init__(
         self,
-        source: str | None = None,
-        destination: str | None = None,
-        headers_only: str | None = None,
+        source: str,
+        destination: str,
+        headers_only: bool,
+    ) -> None: ...
+
+class StreamConfig:
+    def __init__(
+        self,
+        name: str,
+        subjects,
+        max_bytes=None,
+        max_messages=None,
+        max_messages_per_subject=None,
+        discard=None,
+        discard_new_per_subject=None,
+        retention=None,
+        max_consumers=None,
+        max_age=None,
+        max_message_size=None,
+        storage=None,
+        num_replicas=None,
+        no_ack=None,
+        duplicate_window=None,
+        template_owner=None,
+        sealed=None,
+        description=None,
+        allow_rollup=None,
+        deny_delete=None,
+        deny_purge=None,
+        republish=None,
+        allow_direct=None,
+        mirror_direct=None,
+        mirror=None,
+        sources=None,
+        metadata=None,
+        subject_transform=None,
+        compression=None,
+        consumer_limits=None,
+        first_sequence=None,
+        placement=None,
+        persist_mode=None,
+        pause_until=None,
+        allow_message_ttl=None,
+        subject_delete_marker_ttl=None,
+        allow_atomic_publish=None,
+        allow_message_schedules=None,
+        allow_message_counter=None,
     ) -> None: ...

src/exceptions/rust_err.rs

@@ -48,6 +48,8 @@ pub enum NatsrpyError {
     KVUpdateError(#[from] async_nats::jetstream::context::UpdateKeyValueError),
     #[error(transparent)]
     DeleteError(#[from] async_nats::jetstream::kv::DeleteError),
+    #[error(transparent)]
+    CreateStreamError(#[from] async_nats::jetstream::context::CreateStreamError),
 }
 
 impl From<NatsrpyError> for pyo3::PyErr {

src/js/jetstream.rs

@@ -9,7 +9,10 @@ use tokio::sync::RwLock;
 
 use crate::{
     exceptions::rust_err::{NatsrpyError, NatsrpyResult},
-    js::kv::{KVConfig, KeyValue},
+    js::{
+        kv::{KVConfig, KeyValue},
+        stream::StreamConfig,
+    },
     utils::{headers::NatsrpyHeadermapExt, natsrpy_future},
 };
 
@@ -117,15 +120,18 @@ impl JetStream {
         })
     }
 
-    pub fn get_consumer<'py>(
+    pub fn create_stream<'py>(
         &self,
         py: Python<'py>,
-        bucket: String,
+        config: StreamConfig,
     ) -> NatsrpyResult<Bound<'py, PyAny>> {
         let ctx = self.ctx.clone();
         natsrpy_future(py, async move {
             let js = ctx.read().await;
-            Ok(KeyValue::new(js.get_key_value(bucket).await?))
+            Ok(super::stream::Stream::new(
+                js.create_stream(async_nats::jetstream::stream::Config::try_from(config)?)
+                    .await?,
+            ))
         })
     }
 }