Implement Kv/Obj watch.

[sheldygg]

Description

Some updates for NatsBroker:

1. Now you can use just pull_sub=True instead of pull_sub=PullSub() in basic case:

    from faststream import FastStream, Logger
    from faststream.nats import NatsBroker
    
    broker = NatsBroker()
    app = FastStream(broker)
    
    @broker.subscriber("test", stream="stream", pull_sub=True)
    async def handler(msg, logger: Logger):
        logger.info(msg)

2. KeyValue creation and watching API added:

    from faststream import FastStream, Logger
    from faststream.nats import NatsBroker
    
    broker = NatsBroker()
    app = FastStream(broker)
    
    @broker.subscriber("some-key", kv_watch="bucket")
    async def handler(msg: int, logger: Logger):
        logger.info(msg)
    
    @app.after_startup
    async def test():
        kv = await broker.key_value("bucket")
        await kv.put("some-key", b"1")

   Or you can use extended API if it is required

    from faststream.nats import KvWatch
   
    @broker.subscriber("some-key", kv_watch=KvWatch("bucket"))
    async def handler(msg: int, logger: Logger):
        logger.info(msg)

3. ObjectStore API added as well:

   from faststream import FastStream, Logger
   from faststream.nats import NatsBroker
   
   broker = NatsBroker()
   app = FastStream(broker)
   
   @broker.subscriber("file-bucket", obj_watch=True)
   async def handler(filename: str, logger: Logger):
       logger.info(filename)
   
   @app.after_startup
   async def test():
       object_store = await broker.object_storage("file-bucket")
       await object_store.put("some-file.txt", b"1")

   Also, we have built-in ObjectStorage annotation for fastest access to current bucket:

   from faststream.nats.annotations import ObjectStorage
   
   @broker.subscriber("file-bucket", obj_watch=True)
   async def handler(filename: str, logger: Logger, storage: ObjectStorage):
       file = await storage.get(filename)
       logger.info(file.data)

   For sure, we have an extended API as well:

   from faststream.nats import ObjWatch
   
   @broker.subscriber("file-bucket", obj_watch=ObjWatch(include_history=True))
   async def handler(filename: str, logger: Logger):
       logger.info(filename)

Type of change

Please delete options that are not relevant.

• Documentation (typos, code examples, or any documentation updates)
• Bug fix (a non-breaking change that resolves an issue)
• New feature (a non-breaking change that adds functionality)
• Breaking change (a fix or feature that would disrupt existing functionality)
• This change requires a documentation update

Checklist

• My code adheres to the style guidelines of this project (scripts/lint.sh shows no errors)
• I have conducted a self-review of my own code
• I have made the necessary changes to the documentation
• My changes do not generate any new warnings
• I have added tests to validate the effectiveness of my fix or the functionality of my new feature
• Both new and existing unit tests pass successfully on my local environment by running scripts/test-cov.sh
• I have ensured that static analysis tests are passing by running scripts/static-anaylysis.sh
• I have included code examples to illustrate the modifications

[Lancetnik]

• add AsyncAPI tests (KV and OS subscribers should not generate schema)
• add tests for broker.key_value(...) (it must caches buckets)
• add tests for broker.object_storage(...) (it must caches values as well)
• add KeyValue subscriber tests
• add ObjectStorage subscriber tests
• add KeyValue subscriber Path feature tests
• update KeyValue and ObjectStorage documentation
• update FastAPI subscriber annotation
• update NatsRouter annotation