FastAPI integration: Consuming endpoint

[4ydan]

I was wondering if I can solve my usecase with FastStream. I'm trying to have an endpoint which consumes the latest element from a stream and returns it as response.

I have this from the FastAPI documentation but I dont know how to combine it, since right now they both exist concurrently. Also I cant comment out app.include_router(router) otherwise I get

    raise FastAPIError(
fastapi.exceptions.FastAPIError: Prefix and path cannot be both empty (path operation: unknown)

from fastapi import FastAPI
from faststream.redis.fastapi import RedisRouter, Logger
from faststream.redis import StreamSub
from caminterface import Image

router = RedisRouter("redis://localhost:6379")

@router.subscriber(stream=StreamSub("Images"))
async def hello(m: Image, logger: Logger):
    logger.info(m)

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

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

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=5001, log_level="info")

This is what I am looking for.

@router.get("/image")
@router.subscriber(stream=StreamSub("Images"))
async def image_http(m: Image, logger: Logger):
    # process the image do some conversions etc..
    return FileResponse(m, media_type="image/png")

Maybe FastStream is not meant to be used like that? Thx in advance.

[Lancetnik]

@4ydan sorry, it was a bug. I'll fix it with 0.4.3

[4ydan]

Thank you for fixing the bug. So is it possible to consume once on an endpoint call?

[Lancetnik]

With 0.4.3 regular case

@router.subscriber(stream=StreamSub("Images"))
async def hello(m: Image, logger: Logger):
    logger.info(m)

Works fine
I am not sure about what do you mean consume once: if you want to consume the only one message and stop consumer after, you should just raise StopConsume exception

[4ydan]

I want to consume one item from the stream when I call the endpoint /hello for example.

[Lancetnik]

Sorry, FastStream is not designed for such cases. It consumes messages all time without any triggers. In your case should be easely to use redis.xread directly. But I'll think about consumer.getone API, thank you for the idea!

[4ydan]

Thank you, I was thinking that might be the case.