Key-Value Storage#

Overview#

Key-Value storage is just a high-level interface on top of NatsJS.

It is a regular JetStream, where the KV key is a subject.

Put/Update an object to KV by key - it's like publishing a new message to the corresponding subject in the stream.

Thus, the Get command returns not only the current key value but the latest one with an offset of it. Additionally, you can ask for a specific value based on its offset in the KV stream.

This interface provides you with rich abilities to use it like a regular KV storage (ignoring offset) + subscribe to KV key changes + ask for an old KV value revision. So you can use this feature in your application in a really different way. You can find some examples on the NATS developers' official YouTube channel

FastStream Details#

FastStream has some useful methods to help you interact with the Key-Value NATS feature.

First of all, you need to create a Key-Value storage object and put some value to it:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from faststream import FastStream
from faststream.nats import NatsBroker

broker = NatsBroker()
app = FastStream(broker)

@app.after_startup
async def setup_broker():
    key_value = await broker.key_value(bucket="bucket")
    await key_value.put("key", b"Hello!")

Then we are able to use the returned key_value object as a regular NATS one. But, if you want to watch for any changes of some key in the bucket, FastStream allows you to make it via regular @broker.subscriber interface:

1
2
3

@broker.subscriber("key", kv_watch="bucket")
async def handler(msg: str):
    assert msg == "Hello!"

Also, if you want more detailed settings for your Key Value Storage, we have KvWatch object for it:

1
2
3
4
5
6
7
8

from faststream.nats import NatsBroker, KvWatch

@broker.subscriber(
    "key",
    kv_watch=KvWatch("bucket", declare=False),
)
async def handler(msg: str):
    ...