Implemented KV. (#24)

Author: s3rius

python/natsrpy/_natsrpy_rs/js/kv.pyi

@@ -1,13 +1,68 @@
+from datetime import datetime, timedelta
 from typing import final
 
-from natsrpy._natsrpy_rs.js.stream import Placement, Republish, Source, StorageType
 from typing_extensions import Self
 
+from .stream import (
+    Placement,
+    Republish,
+    Source,
+    StorageType,
+    StreamInfo,
+)
+
 __all__ = [
     "KVConfig",
+    "KVEntry",
+    "KVEntryIterator",
+    "KVOperation",
+    "KVStatus",
     "KeyValue",
+    "KeysIterator",
 ]
 
+@final
+class KVStatus:
+    info: StreamInfo
+    bucket: str
+
+@final
+class KVOperation:
+    Put: KVOperation
+    Delete: KVOperation
+    Purge: KVOperation
+
+@final
+class KVEntry:
+    @property
+    def bucket(self) -> str: ...
+    @property
+    def key(self) -> str: ...
+    @property
+    def value(self) -> bytes: ...
+    @property
+    def revision(self) -> int: ...
+    @property
+    def delta(self) -> int: ...
+    @property
+    def created(self) -> datetime: ...
+    @property
+    def operation(self) -> KVOperation: ...
+    @property
+    def seen_current(self) -> bool: ...
+
+@final
+class KVEntryIterator:
+    def __aiter__(self) -> Self: ...
+    async def __anext__(self) -> KVEntry: ...
+    async def next(self, timeout: float | timedelta | None = None) -> KVEntry: ...
+
+@final
+class KeysIterator:
+    def __aiter__(self) -> Self: ...
+    async def __anext__(self) -> str: ...
+    async def next(self, timeout: float | timedelta | None = None) -> str: ...
+
 @final
 class KVConfig:
     """
@@ -63,6 +118,36 @@ class KeyValue:
     def use_jetstream_prefix(self) -> bool: ...
     @property
     def name(self) -> str: ...
-    async def put(self, key: str, value: bytes) -> int: ...
     async def get(self, key: str) -> bytes | None: ...
-    async def delete(self, key: str) -> int: ...
+    async def delete(
+        self,
+        key: str,
+        expect_revision: int | None = None,
+    ) -> int: ...
+    async def update(self, key: str, value: bytes | str, revision: int) -> None: ...
+    async def create(
+        self,
+        key: str,
+        value: bytes | str,
+        ttl: float | timedelta | None = None,
+    ) -> int: ...
+    async def put(self, key: str, value: bytes | str) -> int: ...
+    async def purge(
+        self,
+        key: str,
+        ttl: float | timedelta | None = None,
+        expect_revision: int | None = None,
+    ) -> None: ...
+    async def history(self, key: str) -> KVEntryIterator: ...
+    async def entry(self, key: str, revision: int | None = None) -> KVEntry | None: ...
+    async def watch(
+        self,
+        key: str,
+        from_revision: int | None = None,
+    ) -> KVEntryIterator: ...
+    async def watch_with_history(self, key: str) -> KVEntryIterator: ...
+    async def watch_all(self, from_revision: int | None = None) -> KVEntryIterator: ...
+    async def watch_many(self, keys: list[str]) -> KVEntryIterator: ...
+    async def watch_many_with_history(self, keys: list[str]) -> KVEntryIterator: ...
+    async def keys(self) -> KeysIterator: ...
+    async def status(self) -> KVStatus: ...

python/natsrpy/js.py

@@ -1,15 +1,24 @@
-from ._natsrpy_rs.js import JetStream
+from ._natsrpy_rs.js import JetStream, JetStreamMessage, Publication
 from ._natsrpy_rs.js.consumers import (
     AckPolicy,
     DeliverPolicy,
+    MessagesIterator,
     PriorityPolicy,
     PullConsumer,
     PullConsumerConfig,
     PushConsumer,
     PushConsumerConfig,
     ReplayPolicy,
 )
-from ._natsrpy_rs.js.kv import KeyValue, KVConfig
+from ._natsrpy_rs.js.kv import (
+    KeysIterator,
+    KeyValue,
+    KVConfig,
+    KVEntry,
+    KVEntryIterator,
+    KVOperation,
+    KVStatus,
+)
 from ._natsrpy_rs.js.object_store import (
     ObjectInfo,
     ObjectInfoIterator,
@@ -18,40 +27,55 @@
     ObjectStoreConfig,
 )
 from ._natsrpy_rs.js.stream import (
+    ClusterInfo,
     Compression,
     ConsumerLimits,
     DiscardPolicy,
     External,
+    PeerInfo,
     PersistenceMode,
     Placement,
     Republish,
     RetentionPolicy,
     Source,
+    SourceInfo,
     StorageType,
     Stream,
     StreamConfig,
+    StreamInfo,
     StreamMessage,
+    StreamState,
     SubjectTransform,
 )
 
 __all__ = [
     "AckPolicy",
+    "ClusterInfo",
     "Compression",
     "ConsumerLimits",
     "DeliverPolicy",
     "DiscardPolicy",
     "External",
     "JetStream",
+    "JetStreamMessage",
     "KVConfig",
+    "KVEntry",
+    "KVEntryIterator",
+    "KVOperation",
+    "KVStatus",
     "KeyValue",
+    "KeysIterator",
+    "MessagesIterator",
     "ObjectInfo",
     "ObjectInfoIterator",
     "ObjectLink",
     "ObjectStore",
     "ObjectStoreConfig",
+    "PeerInfo",
     "PersistenceMode",
     "Placement",
     "PriorityPolicy",
+    "Publication",
     "PullConsumer",
     "PullConsumerConfig",
     "PushConsumer",
@@ -60,9 +84,12 @@
     "Republish",
     "RetentionPolicy",
     "Source",
+    "SourceInfo",
     "StorageType",
     "Stream",
     "StreamConfig",
+    "StreamInfo",
     "StreamMessage",
+    "StreamState",
     "SubjectTransform",
 ]

src/exceptions/rust_err.rs

@@ -8,6 +8,8 @@ pub type NatsrpyResult<T> = Result<T, NatsrpyError>;
 pub enum NatsrpyError {
     #[error(transparent)]
     StdIOError(#[from] std::io::Error),
+    #[error(transparent)]
+    UnknownError(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
     #[error("NATS session error: {0}")]
     SessionError(String),
     #[error("Invalid arguemnt: {0}")]
@@ -43,32 +45,40 @@ pub enum NatsrpyError {
     #[error(transparent)]
     UnsubscribeError(#[from] async_nats::UnsubscribeError),
     #[error(transparent)]
-    KeyValueError(#[from] async_nats::jetstream::context::KeyValueError),
+    JSKVError(#[from] async_nats::jetstream::context::KeyValueError),
     #[error(transparent)]
-    CreateKeyValueError(#[from] async_nats::jetstream::context::CreateKeyValueError),
+    JSKVCreateError(#[from] async_nats::jetstream::context::CreateKeyValueError),
     #[error(transparent)]
-    CreateStreamError(#[from] async_nats::jetstream::context::CreateStreamError),
+    JSStreamCreateError(#[from] async_nats::jetstream::context::CreateStreamError),
     #[error(transparent)]
-    GetStreamError(#[from] async_nats::jetstream::context::GetStreamError),
+    JSStreamGetError(#[from] async_nats::jetstream::context::GetStreamError),
     #[error(transparent)]
-    KVUpdateError(#[from] async_nats::jetstream::context::UpdateKeyValueError),
+    JSKVUpdateError(#[from] async_nats::jetstream::context::UpdateKeyValueError),
     #[error(transparent)]
     JSPublishError(#[from] async_nats::jetstream::context::PublishError),
     #[error(transparent)]
+    JSObjectStoreError(#[from] async_nats::jetstream::context::ObjectStoreError),
+    #[error(transparent)]
     KVEntryError(#[from] async_nats::jetstream::kv::EntryError),
     #[error(transparent)]
     KVPutError(#[from] async_nats::jetstream::kv::PutError),
     #[error(transparent)]
-    DeleteError(#[from] async_nats::jetstream::kv::DeleteError),
+    KVUpdateError(#[from] async_nats::jetstream::kv::UpdateError),
+    #[error(transparent)]
+    KVCreateError(#[from] async_nats::jetstream::kv::CreateError),
+    #[error(transparent)]
+    KVWatcherError(#[from] async_nats::jetstream::kv::WatcherError),
+    #[error(transparent)]
+    KVHistoryError(#[from] async_nats::jetstream::kv::HistoryError),
+    #[error(transparent)]
+    KVStatusError(#[from] async_nats::jetstream::kv::StatusError),
     #[error(transparent)]
     StreamDirectGetError(#[from] async_nats::jetstream::stream::DirectGetError),
     #[error(transparent)]
     StreamInfoError(#[from] async_nats::jetstream::stream::InfoError),
     #[error(transparent)]
     StreamPurgeError(#[from] async_nats::jetstream::stream::PurgeError),
     #[error(transparent)]
-    UnknownError(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
-    #[error(transparent)]
     PullMessageError(#[from] async_nats::jetstream::consumer::pull::MessagesError),
     #[error(transparent)]
     ConsumerError(#[from] async_nats::jetstream::stream::ConsumerError),
@@ -81,8 +91,6 @@ pub enum NatsrpyError {
     #[error(transparent)]
     ConsumerUpdateError(#[from] async_nats::jetstream::stream::ConsumerUpdateError),
     #[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),