Added initial object-store implementation. (#12)

Author: s3rius

pyproject.toml

@@ -14,6 +14,9 @@ classifiers = [
   "Topic :: System :: Networking",
 ]
 dynamic = ["version"]
+dependencies = [
+    "typing-extensions>=4.14.0",
+]
 
 [[project.authors]]
 name = "Pavel Kirilin"

python/natsrpy/_natsrpy_rs/js/__init__.pyi

@@ -1,7 +1,7 @@
 from datetime import timedelta
 from typing import Any
 
-from .managers import KVManager, StreamsManager
+from .managers import KVManager, ObjectStoreManager, StreamsManager
 
 class JetStream:
     async def publish(
@@ -17,6 +17,8 @@ class JetStream:
     def kv(self) -> KVManager: ...
     @property
     def streams(self) -> StreamsManager: ...
+    @property
+    def object_store(self) -> ObjectStoreManager: ...
 
 class JetStreamMessage:
     @property

python/natsrpy/_natsrpy_rs/js/managers.pyi

@@ -7,6 +7,7 @@ from .consumers import (
     PushConsumerConfig,
 )
 from .kv import KeyValue, KVConfig
+from .object_store import ObjectStore, ObjectStoreConfig
 from .stream import Stream, StreamConfig
 
 class StreamsManager:
@@ -28,3 +29,8 @@ class ConsumersManager:
     async def create(self, config: PullConsumerConfig) -> PullConsumer: ...
     @overload
     async def create(self, config: PushConsumerConfig) -> PushConsumer: ...
+
+class ObjectStoreManager:
+    async def create(self, config: ObjectStoreConfig) -> ObjectStore: ...
+    async def get(self, bucket: str) -> ObjectStore: ...
+    async def delete(self, bucket: str) -> None: ...

python/natsrpy/_natsrpy_rs/js/object_store.pyi

@@ -0,0 +1,45 @@
+from datetime import timedelta
+
+from typing_extensions import Writer
+
+from .stream import Placement, StorageType
+
+class ObjectStoreConfig:
+    bucket: str
+    description: str | None
+    max_age: timedelta
+    max_bytes: int
+    storage: StorageType
+    num_replicas: int
+    compression: bool
+    placement: Placement | None
+
+    def __init__(
+        self,
+        bucket: str,
+        description: str | None = None,
+        max_age: float | timedelta | None = None,
+        max_bytes: int | None = None,
+        storage: StorageType | None = None,
+        num_replicas: int | None = None,
+        compression: bool | None = None,
+        placement: Placement | None = None,
+    ) -> None: ...
+
+class ObjectStore:
+    async def get(
+        self,
+        name: str,
+        writer: Writer[bytes],
+        chunk_size: int | None = 24576,  # 24MB
+    ) -> None: ...
+    async def put(
+        self,
+        name: str,
+        value: bytes | str,
+        chunk_size: int = 24576,  # 24MB
+        description: str | None = None,
+        headers: dict[str, str | list[str]] | None = None,
+        metadata: dict[str, str] | None = None,
+    ) -> None: ...
+    async def delete(self, name: str) -> None: ...

python/natsrpy/js/__init__.py

@@ -10,6 +10,7 @@
     ReplayPolicy,
 )
 from natsrpy.js.kv import KeyValue, KVConfig
+from natsrpy.js.object_store import ObjectStore, ObjectStoreConfig
 from natsrpy.js.stream import (
     Compression,
     ConsumerLimits,
@@ -37,6 +38,8 @@
     "JetStream",
     "KVConfig",
     "KeyValue",
+    "ObjectStore",
+    "ObjectStoreConfig",
     "PersistenceMode",
     "Placement",
     "PriorityPolicy",

python/natsrpy/js/object_store.py

@@ -0,0 +1,6 @@
+from natsrpy._natsrpy_rs.js.object_store import ObjectStore, ObjectStoreConfig
+
+__all__ = [
+    "ObjectStore",
+    "ObjectStoreConfig",
+]

src/exceptions/rust_err.rs

@@ -6,6 +6,8 @@ pub type NatsrpyResult<T> = Result<T, NatsrpyError>;
 
 #[derive(thiserror::Error, Debug)]
 pub enum NatsrpyError {
+    #[error(transparent)]
+    StdIOError(#[from] std::io::Error),
     #[error("NATS session error: {0}")]
     SessionError(String),
     #[error("Invalid arguemnt: {0}")]
@@ -72,6 +74,14 @@ pub enum NatsrpyError {
     PushConsumerMessageError(#[from] async_nats::jetstream::consumer::push::MessagesError),
     #[error(transparent)]
     ConsumerStreamError(#[from] async_nats::jetstream::consumer::StreamError),
+    #[error(transparent)]
+    ObjectStoreError(#[from] async_nats::jetstream::context::ObjectStoreError),
+    #[error(transparent)]
+    ObjectStoreGetError(#[from] async_nats::jetstream::object_store::GetError),
+    #[error(transparent)]
+    ObjectStorePutError(#[from] async_nats::jetstream::object_store::PutError),
+    #[error(transparent)]
+    ObjectStoreDeleteError(#[from] async_nats::jetstream::object_store::DeleteError),
 }
 
 impl From<NatsrpyError> for pyo3::PyErr {

src/js/counters.rs

@@ -6,7 +6,7 @@ use tokio::sync::RwLock;
 
 use crate::{
     exceptions::rust_err::{NatsrpyError, NatsrpyResult},
-    js::managers::{kv::KVManager, streams::StreamsManager},
+    js::managers::{kv::KVManager, object_store::ObjectStoreManager, streams::StreamsManager},
     utils::{headers::NatsrpyHeadermapExt, natsrpy_future, py_types::SendableValue},
 };
 
@@ -26,6 +26,24 @@ impl JetStream {
 
 #[pyo3::pymethods]
 impl JetStream {
+    #[getter]
+    #[must_use]
+    pub fn kv(&self) -> KVManager {
+        KVManager::new(self.ctx.clone())
+    }
+
+    #[getter]
+    #[must_use]
+    pub fn streams(&self) -> StreamsManager {
+        StreamsManager::new(self.ctx.clone())
+    }
+
+    #[getter]
+    #[must_use]
+    pub fn object_store(&self) -> ObjectStoreManager {
+        ObjectStoreManager::new(self.ctx.clone())
+    }
+
     #[pyo3(signature = (
         subject,
         payload,
@@ -66,16 +84,4 @@ impl JetStream {
             Ok(())
         })
     }
-
-    #[getter]
-    #[must_use]
-    pub fn kv(&self) -> KVManager {
-        KVManager::new(self.ctx.clone())
-    }
-
-    #[getter]
-    #[must_use]
-    pub fn streams(&self) -> StreamsManager {
-        StreamsManager::new(self.ctx.clone())
-    }
 }

src/js/jetstream.rs

@@ -1,5 +1,6 @@
 pub mod consumers;
 pub mod kv;
+pub mod object_store;
 pub mod streams;
 
 #[pyo3::pymodule(submodule, name = "managers")]
@@ -9,5 +10,7 @@ pub mod pymod {
     #[pymodule_export]
     use super::kv::KVManager;
     #[pymodule_export]
+    use super::object_store::ObjectStoreManager;
+    #[pymodule_export]
     use super::streams::StreamsManager;
 }