Skip to content

Object Storage#

Object storage is almost identical to the Key-Value storage concept, so you can reuse the guide.

Overview#

Object Storage is just a high-level interface on top of NatsJS.

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

The main difference between KV and Object storages is that in the Object storage, you can store files greater than 1MB (a limitation of KV). It has no limit on the maximum object size and stores it in chunks (each message is an object chunk), so you can literally stream huge objects through NATS.

FastStream Details#

FastStream has some useful methods to help you interact with the Object Storage NATS feature.

First of all, you need to create an Object Storage object and put some value to it:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from io import BytesIO

from faststream import FastStream
from faststream.nats import NatsBroker

broker = NatsBroker()
app = FastStream(broker)

@app.after_startup
async def test_send():
    object_storage = await broker.object_storage("example-bucket")
    await object_storage.put("file.txt", BytesIO(b"File mock"))

Tip

  • BytesIO - is a Readable object used to emulate a file opened for reading.

  • broker.object_storage(bucket="example-bucket") is an idempotent method. It means that it stores all already created storages in memory and does not make a new request to NATS if you are trying to call it for the same bucket.

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

1
2
3
@broker.subscriber("example-bucket", obj_watch=True)
async def handler(filename: str):
    assert filename == "file.txt"

NATS delivers you just a filename (and some more metainformation you can access via message.raw_message) because files can be any size. The framework should protect your service from memory overflow, so we can't load the whole file content right into memory. But you can do it manually the following way:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream.nats.annotations import ObjectStorage

@broker.subscriber("example-bucket", obj_watch=True)
async def handler(
    filename: str,
    storage: ObjectStorage,
    logger: Logger,
):
    assert filename == "file.txt"
    file = await storage.get(filename)
    logger.info(file.data)

Note

faststream.nats.annotations.ObjectStorage is your current bucket, so you don't need to put it into the context manually.

Also, if you want more detailed settings for your Object Storage, we have ObjWatch object for it:

1
2
3
4
5
6
7
8
from faststream.nats import NatsBroker, ObjWatch

@broker.subscriber(
    "example-bucket",
    obj_watch=ObjWatch(declare=False),
)
async def handler(filename: str):
    ...