Coverage for examples / nats / e07_object_storage.py: 80%

1from io import BytesIO (empty)

2 

3from faststream import FastStream, Logger (empty)

4from faststream.nats import NatsBroker (empty)

5from faststream.nats.annotations import ObjectStorage (empty)

6 

7broker = NatsBroker() (empty)

8app = FastStream(broker) (empty)

9 

10 

11@broker.subscriber("example-bucket", obj_watch=True) (empty)

12async def handler(filename: str, storage: ObjectStorage, logger: Logger) -> None: (empty)

13 assert filename == "file.txt" (empty)

14 file = await storage.get(filename) (empty)

15 logger.info(file.data) (empty)

16 

17 

18@app.after_startup (empty)

19async def test_send() -> None: (empty)

20 os = await broker.object_storage("example-bucket") 

21 await os.put("file.txt", BytesIO(b"File mock"))