aiomqtt-router

[badge]

Hi! I just discovered aiomqtt yesterday and it looks great, almost exactly what I need. The missing part was that I have to handle quite a lot of subscriptions, and wanted an easy way to do that without writing loads of boilerplate/losing my mind.

The result is aiomqtt-router, which enables decorator-based subscription handlers with both regular functions and coroutines:

import asyncio
import aiomqtt

from aiomqtt_router import AiomqttRouter

router = AiomqttRouter()


@router.subscribe("humidity/#")
def handle_humidity(message: aiomqtt.Message):
    print(f"[humidity/#] {message.payload}")


@router.subscribe("temperature/#")
async def handle_temperature(message: aiomqtt.Message):
    print(f"[temperature/#] {message.payload}")


async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await router(client)


if __name__ == "__main__":
    asyncio.run(main())

It's pretty barebones right now and has zero automated tests, but I thought it might be of interest. :)

[frederikaalund]

Excellent idea! 👍 Personally, I think this is a really interesting direction. I see aiomqtt-router as the "FastAPI of aiomqtt". :) Just like FastAPI uses, e.g., uvicorn underneath, then aiomqtt-router uses aiomqtt underneath. I think it makes a lot of sense to distinguish between the high-level aiomqtt-router and the low-level aiomqtt. Makes it easier for each package to "do one thing and to do it well".

Makes me think of #287 and whether it should go into one or the other package. :)

Keep up the good work! 👍