Added otlp instrumentation.

Author: s3rius

.gitignore

@@ -1,4 +1,5 @@
 /target
+delete-me-*
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

.pre-commit-config.yaml

@@ -18,7 +18,7 @@ repos:
         name: python mypy
         always_run: true
         pass_filenames: false
-        args: ["python"]
+        args: ["python", "examples"]
   - repo: https://github.com/astral-sh/ruff-pre-commit
     rev: v0.15.7
     hooks:

Cargo.toml

@@ -18,7 +18,7 @@ async-nats = "0.46"
 bytes = "1.11.1"
 futures-util = "0.3.32"
 log = "0.4.29"
-pyo3 = { version = "0.28", features = ["experimental-inspect"] }
+pyo3 = { version = "0.28", features = ["abi3", "experimental-inspect"] }
 pyo3-async-runtimes = { version = "0.28", features = ["tokio-runtime"] }
 pyo3-log = "0.13.3"
 serde = { version = "1.0.228", features = ["derive"] }

examples/consumers.py

@@ -44,13 +44,13 @@ async def main() -> None:
     # We use messages() to get async iterator which we
     # use to get messages for push_consumer.
     async for push_message in await push_consumer.messages():
-        print(f"[FROM_PUSH] {push_message.payload}")  # noqa: T201
+        print(f"[FROM_PUSH] {push_message.payload!r}")  # noqa: T201
         await push_message.ack()
         break
 
     # Pull consumers have to request batches of messages.
     for pull_message in await pull_consumer.fetch(max_messages=10):
-        print(f"[FROM_PULL] {pull_message.payload}")  # noqa: T201
+        print(f"[FROM_PULL] {pull_message.payload!r}")  # noqa: T201
         await pull_message.ack()
 
     # Cleanup

examples/kv.py

@@ -30,8 +30,6 @@ async def main() -> None:
 
     await kv.delete("test-key")
 
-    # Alternatively you can
-    # use await watcher.next()
     async for event in watcher:
         print("[EVENT]", event)  # noqa: T201
         break

examples/request_reply.py

@@ -12,11 +12,11 @@ async def main() -> None:
     # Here we create responder, that will be
     # answering to our requests.
     async def responder(message: Message) -> None:
-        print(f"[REQUEST]: {message.payload}, headers={message.headers}")  # noqa: T201
+        print(f"[REQUEST]: {message.payload!r}, headers={message.headers}")  # noqa: T201
         if message.reply:
             await nats.publish(
                 message.reply,
-                f"reply to {message.payload}",
+                f"reply to {message.payload!r}",
                 headers=message.headers,
             )
 

examples/subscriptions.py

@@ -11,7 +11,7 @@ async def main() -> None:
     cb_lock = asyncio.Event()
 
     async def callback(message: Message) -> None:
-        print(f"[FROM_CALLBACK] {message.payload}")  # noqa: T201
+        print(f"[FROM_CALLBACK] {message.payload!r}")  # noqa: T201
         cb_lock.set()
 
     # When subscribing you can set callback.
@@ -21,8 +21,6 @@ async def callback(message: Message) -> None:
 
     # When callback is not set, you get a subscription
     # that should be used along with `async for`
-    # loop, or alternatively you can call
-    # `await iter_sub.next()` to get a single message.
     iter_sub = await nats.subscribe("iter-subj")
 
     # Subscriptions with queue argument create
@@ -40,10 +38,10 @@ async def callback(message: Message) -> None:
     await queue_sub.unsubscribe(limit=1)
 
     async for message in iter_sub:
-        print(f"[FROM_ITERATOR] {message.payload}")  # noqa: T201
+        print(f"[FROM_ITERATOR] {message.payload!r}")  # noqa: T201
 
     async for message in queue_sub:
-        print(f"[FROM_QUEUED] {message.payload}")  # noqa: T201
+        print(f"[FROM_QUEUED] {message.payload!r}")  # noqa: T201
 
     # Making sure that the message in callback is received.
     await cb_lock.wait()

pyproject.toml

@@ -23,6 +23,16 @@ dynamic = ["version"]
 name = "Pavel Kirilin"
 email = "s3riussan@gmail.com"
 
+[project.entry-points.opentelemetry_instrumentor]
+natsrpy = "natsrpy.instrumentation:NatsrpyInstrumentor"
+
+[project.optional-dependencies]
+opentelemetry = [
+  "opentelemetry-api (>=1.38.0,<2.0.0)",
+  "opentelemetry-instrumentation (>=0.59b0,<1)",
+  "opentelemetry-semantic-conventions (>=0.59b0,<1)",
+]
+
 [dependency-groups]
 dev = [
   "anyio>=4,<5",
@@ -38,9 +48,6 @@ dev = [
 requires = ["maturin>=1.12,<2.0"]
 build-backend = "maturin"
 
-[tool.uv]
-package = false
-
 [tool.maturin]
 bindings = "pyo3"
 features = ["pyo3/extension-module"]
@@ -55,6 +62,7 @@ packages = ["natsrpy"]
 pretty = true
 implicit_reexport = true
 allow_untyped_decorators = true
+namespace_packages = true
 warn_return_any = false
 
 [tool.pytest]
@@ -128,3 +136,6 @@ ignore-decorators = ["typing.overload"]
 
 [tool.ruff.lint.pylint]
 allow-magic-value-types = ["int", "str", "float"]
+
+[tool.uv]
+package = false

python/natsrpy/_natsrpy_rs/__init__.pyi

@@ -33,27 +33,12 @@ class Message:
     description: str | None
     length: int
 
+    def __len__(self) -> int: ...
+
 @final
 class IteratorSubscription:
-    """Async iterator subscription for receiving NATS messages.
-
-    Returned by :meth:`Nats.subscribe` when no callback is provided.
-    Messages can be received using ``async for`` or by calling :meth:`next`
-    directly.
-    """
-
-    def __aiter__(self) -> IteratorSubscription: ...
+    def __aiter__(self) -> Self: ...
     def __anext__(self) -> Future[Message]: ...
-    def next(self, timeout: float | timedelta | None = None) -> Future[Message]:
-        """Receive the next message from the subscription.
-
-        :param timeout: maximum time to wait for a message in seconds
-            or as a timedelta, defaults to None (wait indefinitely).
-        :return: the next message.
-        :raises StopAsyncIteration: when the subscription is drained or
-            unsubscribed.
-        """
-
     def unsubscribe(self, limit: int | None = None) -> Future[None]:
         """Unsubscribe from the subject.
 
@@ -96,6 +81,22 @@ class Nats:
     access over a connection to one or more NATS servers.
     """
 
+    @property
+    def addr(self) -> list[str]: ...
+    @property
+    def user_and_pass(self) -> tuple[str, str]: ...
+    @property
+    def nkey(self) -> str | None: ...
+    @property
+    def token(self) -> str | None: ...
+    @property
+    def custom_inbox_prefix(self) -> str | None: ...
+    @property
+    def read_buffer_capacity(self) -> int: ...
+    @property
+    def sender_capacity(self) -> int: ...
+    @property
+    def max_reconnects(self) -> int | None: ...
     def __new__(
         cls,
         /,

python/natsrpy/_natsrpy_rs/js/__init__.pyi

@@ -159,6 +159,10 @@ class JetStreamMessage:
     def token(self) -> str | None:
         """Authentication token, if applicable."""
 
+    @property
+    def length(self) -> int:
+        """Message's payload length."""
+
     def ack(self, double: bool = False) -> Future[None]:
         """Acknowledge that a message was handled.
 
@@ -208,3 +212,5 @@ class JetStreamMessage:
 
         :param double: whether to wait for server response, defaults to False.
         """
+
+    def __len__(self) -> int: ...