natsext 0.4.0

Core NATS Extensions is a set of utilities providing additional features to Core NATS component of nats-py client.

Core NATS Extensions

Core NATS Extensions is a set of utilities providing additional features to Core NATS component of nats-py client.

Installation

uv add natsext

Utilities

see examples.py for a runnable version of all snippets below.

request_many

request_many is a utility that allows you to send a single request and await multiple responses. This allows you to implement various patterns like scatter-gather or streaming responses.

Responses are returned in an async iterator, which you can iterate over to receive messages. When a termination condition is met, the iterator is closed (and no error is returned).

import nats
import natsext

nc = await nats.connect()

# Basic usage
async for msg in natsext.request_many(nc, "subject", b"request data"):
    print(f"Received: {msg.data}")

Alternatively, use request_many_msg to send a Msg request:

import nats
from nats.aio.msg import Msg
import natsext

nc = await nats.connect()

msg = Msg(
    nc,
    subject="subject",
    data=b"request data",
    headers={
        "Key": "Value",
    },
)
async for response in natsext.request_many_msg(nc, msg):
    print(f"Received: {response.data}")

Configuration

You can configure the following options:

• timeout: Overall timeout for the request operation (float, seconds)
• stall: Stall timer, useful in scatter-gather scenarios where subsequent responses are expected within a certain timeframe (float, seconds)
• max_messages: Maximum number of messages to receive (int)
• sentinel: Function that stops returning responses once it returns True for a message (Callable[[Msg], bool])

import nats
import natsext

nc = await nats.connect()

# With all options
async for msg in natsext.request_many(
    nc,
    "subject",
    b"request data",
    timeout=5.0,
    stall=0.1,
    max_messages=3,
    sentinel=None,  # Don't use sentinel here to show max_messages working
):
    print(f"Received: {msg.data}")

Default Sentinel

The package includes a default_sentinel function that stops receiving messages once a message with an empty payload is received:

import nats
import natsext

nc = await nats.connect()

async for msg in natsext.request_many(
    nc, "subject", b"request", sentinel=natsext.default_sentinel
):
    print(f"Received: {msg.data}")