FastAPI / NATS: Batching, connection handling w/ router.broker.publish()

[sphr2k]

Hi everyone,

I use FastStream w/ NATS (JetStream) and the FastAPI integration. My goal is to process a batch of messages and publish individual responses for each message in the batch.

Unless I'm overlooking sth, with batching, using the Publisher Decorator I can only send a message for the entire batch, i.e.

js = JStream(name="stream")

pull_sub = PullSub(
    batch_size=2,
    batch=True,
)

@router.subscriber("in", stream=js, pull_sub=pull_sub)
@router.publisher("out")
async def test_handler(msg: list[str]):
    print(msg)
    return msg

will yield

$ nats sub out

[#1] Received on "out"
content-type: application/json
correlation_id: a48d2ff2-37ff-43e5-890c-9472bfc63c60

["msg1", "msg2"]

To get around this, I guess only direct publishing can be used.

@router.subscriber("in", stream=js, pull_sub=pull_sub)
async def test_handler(msg: list[str]):
    print(msg)
    await router.broker.publish(msg, "out")

However, this throws an error:

AssertionError: Please, `connect()` the broker first

Adding a manual

await router.broker.connect()

does work, but is this how it's supposed to be?

Thanks a lot!

[Lancetnik]

Please, do not use router.broker directly. The preferred way for all cases - getting broker from the Context

[csessh]

I'm curious and if you don't mind me asking @Lancetnik , what's the downside to directly using router.broker ? anti pattern?

[Lancetnik]

As user you can't be sure, that you are using broker (or right broker object) in a right moment. Using broker from the context you always has guarantee that is connected and already running one.