wip

Author: caspervonb

nats-jetstream/examples/pull_consume_callback.py

@@ -43,7 +43,9 @@ async def main():
     s = await js.create_stream(name="TEST_STREAM", subjects=["FOO.*"])
 
     cons = await s.upsert_consumer(
-        name="TestConsumerConsume", durable_name="TestConsumerConsume", ack_policy="explicit"
+        name="TestConsumerConsume",
+        durable_name="TestConsumerConsume",
+        ack_policy="explicit"
     )
 
     publish_task = asyncio.create_task(endless_publish(nc, js))

nats-jetstream/examples/pull_consume_iterator.py

@@ -30,7 +30,9 @@ async def main():
     s = await js.create_stream(name="TEST_STREAM", subjects=["FOO.*"])
 
     cons = await s.upsert_consumer(
-        name="TestConsumerIterator", durable_name="TestConsumerIterator", ack_policy="explicit"
+        name="TestConsumerIterator",
+        durable_name="TestConsumerIterator",
+        ack_policy="explicit"
     )
 
     publish_task = asyncio.create_task(endless_publish(nc, js))

nats-jetstream/examples/pull_fetch.py

@@ -29,7 +29,9 @@ async def main():
     s = await js.create_stream(name="TEST_STREAM", subjects=["FOO.*"])
 
     cons = await s.upsert_consumer(
-        name="TestConsumerListener", durable_name="TestConsumerListener", ack_policy="explicit"
+        name="TestConsumerListener",
+        durable_name="TestConsumerListener",
+        ack_policy="explicit"
     )
 
     publish_task = asyncio.create_task(endless_publish(nc, js))

nats-jetstream/src/nats/jetstream/__init__.py

@@ -107,7 +107,10 @@ def from_response(cls, data: api.AccountInfo) -> AccountInfo:
             limits=AccountLimits.from_response(data["limits"]),
             api=APIStats.from_response(data["api"]),
             domain=data.get("domain"),
-            tiers={k: Tier.from_response(v) for k, v in data["tiers"].items()} if "tiers" in data else None,
+            tiers={
+                k: Tier.from_response(v)
+                for k, v in data["tiers"].items()
+            } if "tiers" in data else None,
         )
 
 
@@ -133,7 +136,12 @@ def from_response(cls, data: api.PublishAck) -> PublishAck:
 class JetStream:
     """JetStream context."""
 
-    def __init__(self, client: NatsClient, prefix: str = "$JS.API", domain: str | None = None) -> None:
+    def __init__(
+        self,
+        client: NatsClient,
+        prefix: str = "$JS.API",
+        domain: str | None = None
+    ) -> None:
         """Initialize JetStream client.
 
         Args:
@@ -185,7 +193,8 @@ async def publish(
 
         return publish_ack
 
-    async def stream_names(self, subject: str | None = None) -> AsyncIterator[str]:
+    async def stream_names(self,
+                           subject: str | None = None) -> AsyncIterator[str]:
         """Get an async iterator over all stream names.
 
         Args:
@@ -197,7 +206,9 @@ async def stream_names(self, subject: str | None = None) -> AsyncIterator[str]:
         offset = 0
         total = None
         while True:
-            response = await self._api.stream_names(offset=offset, subject=subject)
+            response = await self._api.stream_names(
+                offset=offset, subject=subject
+            )
             streams = response.get("streams", [])
             if streams is None:
                 streams = []
@@ -216,7 +227,9 @@ async def stream_names(self, subject: str | None = None) -> AsyncIterator[str]:
             # Increment offset by the number of streams we received
             offset += len(streams)
 
-    async def list_streams(self, subject: str | None = None) -> AsyncIterator[StreamInfo]:
+    async def list_streams(self,
+                           subject: str | None = None
+                           ) -> AsyncIterator[StreamInfo]:
         """Get an async iterator over all streams.
 
         Args:
@@ -228,7 +241,9 @@ async def list_streams(self, subject: str | None = None) -> AsyncIterator[Stream
         offset = 0
         total = None
         while True:
-            response = await self._api.stream_list(offset=offset, subject=subject)
+            response = await self._api.stream_list(
+                offset=offset, subject=subject
+            )
             streams = response.get("streams", [])
             if streams is None:
                 streams = []
@@ -285,7 +300,13 @@ async def get_stream(self, name: str) -> Stream:
         info = await self.get_stream_info(name)
         return Stream(self, name, info)
 
-    async def create_consumer(self, stream_name: str, name: str, durable_name: str | None = None, **config) -> Consumer:
+    async def create_consumer(
+        self,
+        stream_name: str,
+        name: str,
+        durable_name: str | None = None,
+        **config
+    ) -> Consumer:
         """Create a consumer for a stream.
 
         Args:
@@ -309,7 +330,9 @@ async def create_consumer(self, stream_name: str, name: str, durable_name: str |
         # Create the consumer via the stream
         return await stream.create_consumer(name, **consumer_config)
 
-    async def get_consumer(self, stream_name: str, consumer_name: str) -> Consumer:
+    async def get_consumer(
+        self, stream_name: str, consumer_name: str
+    ) -> Consumer:
         """Get a consumer by name.
 
         Args:
@@ -322,7 +345,9 @@ async def get_consumer(self, stream_name: str, consumer_name: str) -> Consumer:
         stream = await self.get_stream(stream_name)
         return await stream.get_consumer(consumer_name)
 
-    async def delete_consumer(self, stream_name: str, consumer_name: str) -> bool:
+    async def delete_consumer(
+        self, stream_name: str, consumer_name: str
+    ) -> bool:
         """Delete a consumer.
 
         Args:
@@ -335,7 +360,9 @@ async def delete_consumer(self, stream_name: str, consumer_name: str) -> bool:
         stream = await self.get_stream(stream_name)
         return await stream.delete_consumer(consumer_name)
 
-    async def update_consumer(self, stream_name: str, consumer_name: str, **config) -> Consumer:
+    async def update_consumer(
+        self, stream_name: str, consumer_name: str, **config
+    ) -> Consumer:
         """Update a consumer.
 
         Args:
@@ -362,7 +389,9 @@ async def consumer_names(self, stream_name: str) -> AsyncIterator[str]:
         total = None
 
         while True:
-            response = await self._api.consumer_list(stream_name, offset=offset)
+            response = await self._api.consumer_list(
+                stream_name, offset=offset
+            )
             consumers = response.get("consumers", [])
 
             if consumers is None:
@@ -382,7 +411,8 @@ async def consumer_names(self, stream_name: str) -> AsyncIterator[str]:
             # Increment offset
             offset += len(consumers)
 
-    async def consumers_info(self, stream_name: str) -> AsyncIterator[ConsumerInfo]:
+    async def consumers_info(self,
+                             stream_name: str) -> AsyncIterator[ConsumerInfo]:
         """Get an async iterator over all consumer info objects for a stream.
 
         Args:
@@ -395,7 +425,9 @@ async def consumers_info(self, stream_name: str) -> AsyncIterator[ConsumerInfo]:
         total = None
 
         while True:
-            response = await self._api.consumer_names(stream_name, offset=offset)
+            response = await self._api.consumer_names(
+                stream_name, offset=offset
+            )
             consumers = response.get("consumers", [])
 
             if consumers is None:
@@ -416,7 +448,9 @@ async def consumers_info(self, stream_name: str) -> AsyncIterator[ConsumerInfo]:
             # Increment offset
             offset += len(consumers)
 
-    async def get_consumer_info(self, stream_name: str, consumer_name: str) -> ConsumerInfo:
+    async def get_consumer_info(
+        self, stream_name: str, consumer_name: str
+    ) -> ConsumerInfo:
         """Get consumer info.
 
         Args:
@@ -467,11 +501,15 @@ async def get_message(self, stream: str, sequence: int) -> StreamMessage:
             subject=message["subject"],
             sequence=message["seq"],
             data=data or b"",
-            time=datetime.fromisoformat(message["time"].replace("Z", "+00:00")),
+            time=datetime.fromisoformat(
+                message["time"].replace("Z", "+00:00")
+            ),
             headers=headers,
         )
 
-    async def get_last_message_for_subject(self, stream: str, subject: str) -> StreamMessage:
+    async def get_last_message_for_subject(
+        self, stream: str, subject: str
+    ) -> StreamMessage:
         """Get the last message for a subject directly from a stream.
 
         This is a direct message get that requires the stream to have allow_direct=true.
@@ -504,12 +542,18 @@ async def get_last_message_for_subject(self, stream: str, subject: str) -> Strea
             subject=message["subject"],
             sequence=message["seq"],
             data=data or b"",
-            time=datetime.fromisoformat(message["time"].replace("Z", "+00:00")),
+            time=datetime.fromisoformat(
+                message["time"].replace("Z", "+00:00")
+            ),
             headers=headers,
         )
 
 
-def new(client: NatsClient, prefix: str = "$JS.API", domain: str | None = None) -> JetStream:
+def new(
+    client: NatsClient,
+    prefix: str = "$JS.API",
+    domain: str | None = None
+) -> JetStream:
     """Create a new JetStream instance.
 
     Args: