Suggestion: Litestar Plugin

[hasansezertasan]

Currently, FastStream can be used as part of a FastAPI application. What do you think about supporting another framework like Litestar?

[Lancetnik]

@hasansezertasan probably we can. But can you suggest any API how it should looks like?

[hasansezertasan]

Gladly, my suggestion below is based on this example:

import uvicorn
from faststream.rabbit.litestar import (
    Logger,
    RabbitBroker,
    RabbitConfig,
    RabbitPlugin,
    publisher,
    subscriber,
)
from litestar import Litestar, get


# Define the handler
@publisher
@subscriber("test-queue")
async def handler(name: str, logger: Logger) -> str:
    message = f"Hello, {name}!"
    logger.info(message)
    return f"{message} created"


# Define the endpoint
@get("/{name:str}")
async def index(name: str, broker: RabbitBroker) -> str:
    message = f"Hello, {name}!"
    await broker.publish(message=message, queue="test-queue")
    return message


# Assamble the app
app = Litestar(
    plugins=[RabbitPlugin(config=RabbitConfig("amqp://guest:guest@localhost:5672/"))],
    route_handlers=[index],
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

When I think about out-of-the-box Litestar integration, I see this 🤓. I expect to have auto-generated endpoints for AsyncAPI, Like FastAPI Integration.

[Lancetnik]

@hasansezertasan looks great, but how handler endpoint should be wired by RabbitPlugin? We are passing http index endpoint to the app, but I don't see such mechanism for FastStream endpoints

What do you think about following design?

import uvicorn
from faststream.rabbit.litestar import RabbitBroker
from faststream.litestar import Logger, BrokerPlugin
from litestar import Litestar

# regular FastStream broker
broker = RabbitBroker()

@broker.publisher("out-queue")
@broker.subscriber("test-queue")
async def handler(name: str, logger: Logger) -> str:
    message = f"Hello, {name}!"
    logger.info(message)
    return f"{message} created"

# Assamble the app
app = Litestar(
    plugins=[BrokerPlugin(broker)],
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

[hasansezertasan]

@hasansezertasan looks great, but how handler endpoint should be wired by RabbitPlugin? We are passing http index endpoint to the app, but I don't see such mechanism for FastStream endpoints

Thank you 🤓!

It seems that I forgot to do the function registration when initializing the plugin. Here is what I was really thinking about:

import uvicorn
from faststream.rabbit.litestar import (
    Logger,
    RabbitBroker,
    RabbitConfig,
    RabbitPlugin,
    publisher,
    subscriber,
)
from litestar import Litestar, get


# Define the handler
@publisher
@subscriber("test-queue")
async def handler(name: str, logger: Logger) -> str:
    message = f"Hello, {name}!"
    logger.info(message)
    return f"{message} created"


# Define the endpoint
@get("/{name:str}")
async def index(name: str, broker: RabbitBroker) -> str:
    message = f"Hello, {name}!"
    await broker.publish(message=message, queue="test-queue")
    return message


# Assamble the app
app = Litestar(
    plugins=[
        RabbitPlugin(
            config=RabbitConfig("amqp://guest:guest@localhost:5672/"),
            handlers=[handler],
        ),
    ],
    route_handlers=[index],
)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

What do you think about following design?

Your approach seems convenient.

[sobolevn]

I would advocate for community plugin for Litestar. The closest match would be https://github.com/taskiq-python/taskiq-litestar

I don't see any reasons why this should be bundled with the core. From both FS and Litestar sides.