Send Image data in the queue

[Dronakurl]

Hi, I am new to faststream.
Is it possible to send image data from and to faststream routes or does it have to be serializable data?
The workaround would be to save the images in the filesystem and send and recieve the path. Is this the way?

[Lancetnik]

Sure, you can send raw bytes with FastStream the following way

@broker.subscriber(...)
async def handler(file: bytes):
    ....

await broker.publish(b"file_content", ....)

But you should be accurate with your files size. Probably, you should take a look at NATS Object Storage feature or upload your files to S3 and subscribe on its events

[Dronakurl]

Thank you very much!
But I get "2024-05-17 01:05:39,459 ERROR - binary | 1-17159007 - UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 0: invalid start byte"
I could not find anything useful to debug this error. Do you have an idea what to do, please?

Here is my code, which should just take a picture from the webcam and respond on a fastapi endpoint

import cv2
import uvicorn
from fastapi import FastAPI, Response
from faststream.kafka.fastapi import KafkaRouter, Logger

router = KafkaRouter("localhost:9092")
vid = cv2.VideoCapture(0)


@router.subscriber("binary")
@router.get("/", response_class=Response)
async def bin(mm: bytes, logger: Logger):
    print(len(mm))
    print(type(mm))
    x = Response(content=mm, media_type="image/jpeg")
    return x


@router.after_startup
async def publish(app: FastAPI):
    await asyncio.sleep(2)
    _, f = vid.read()
    _, buffer = cv2.imencode(".jpg", f)
    bb = buffer.tobytes()
    cv2.imshow("xx", f)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        cv2.destroyAllWindows()
    print(len(bb))
    await router.broker.publish(bb, "binary")


app = FastAPI(lifespan=router.lifespan_context)
app.include_router(router)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=False, workers=1, log_level="warning")

[sheldygg]

Hello, I can suggest you to use NatsBroker, it has a storage object in which you can easily store your files as bytes.
Nats is supported by FastStream and it can be easily used.

And further, PR #1383 with updating Nats KV, OBJ storages are coming.

[Dronakurl]

Apparently, transferring bytes works in faststream, but not with the fastapi plugin.
When adapting the example on the webpage to bytes output, it raises the UnicodeDecodeError

from pathlib import Path

from fastapi import Depends, FastAPI
from faststream.kafka.fastapi import KafkaRouter, Logger
from pydantic import BaseModel

router = KafkaRouter("localhost:9092")


class Incoming(BaseModel):
    m: dict


def call():
    return True


@router.subscriber("test")
@router.publisher("response")
async def hello(m: Incoming, logger: Logger, d=Depends(call)):
    logger.info(m)
    # >> This works
    # return "hUHU"
    # >> This doesn't -> decoding error
    return Path("tt.jpg").read_bytes()


@router.after_startup
async def test(app: FastAPI):
    await router.broker.publish(Incoming(m=dict()), "test")


@router.get("/")
async def hello_http():
    return "Hello, HTTP!"


app = FastAPI(lifespan=router.lifespan_context)
app.include_router(router)