fastapi-mqtt 1.2.0

fastapi-mqtt is extension for MQTT protocol

fastapi-mqtt

MQTT is a lightweight publish/subscribe messaging protocol designed for M2M (machine to machine) telemetry in low bandwidth environments. Fastapi-mqtt is the client for working with MQTT.

For more information about MQTT, please refer to here: MQTT

Fatapi-mqtt wraps around gmqtt module. Gmqtt Python async client for MQTT client implementation. Module has support of MQTT version 5.0 protocol

---

Documentation: FastApi-MQTT

The key feature are:

MQTT specification avaliable with help decarator methods using callbacks:

• on_connect()

• on_disconnect()

• on_subscribe()

• on_message()

• subscribe(topic)

• MQTT Settings available with pydantic class

• Authentication to broker with credentials

• unsubscribe certain topics and publish to certain topics

🔨 Installation

 $ pip install fastapi-mqtt

🕹 Guide

from fastapi import FastAPI
from fastapi_mqtt import FastMQTT, MQTTConfig

app = FastAPI()

mqtt_config = MQTTConfig()

mqtt = FastMQTT(
    config=mqtt_config
)

mqtt.init_app(app)



@mqtt.on_connect()
def connect(client, flags, rc, properties):
    mqtt.client.subscribe("/mqtt") #subscribing mqtt topic
    print("Connected: ", client, flags, rc, properties)

@mqtt.on_message()
async def message(client, topic, payload, qos, properties):
    print("Received message: ",topic, payload.decode(), qos, properties)

@mqtt.subscribe("my/mqtt/topic/#")
async def message_to_topic(client, topic, payload, qos, properties):
    print("Received message to specific topic: ", topic, payload.decode(), qos, properties)

@mqtt.subscribe("my/mqtt/topic/#", qos=2)
async def message_to_topic_with_high_qos(client, topic, payload, qos, properties):
    print("Received message to specific topic and QoS=2: ", topic, payload.decode(), qos, properties)

@mqtt.on_disconnect()
def disconnect(client, packet, exc=None):
    print("Disconnected")

@mqtt.on_subscribe()
def subscribe(client, mid, qos, properties):
    print("subscribed", client, mid, qos, properties)

Publish method:

async def func():
    mqtt.publish("/mqtt", "Hello from Fastapi") #publishing mqtt topic

    return {"result": True,"message":"Published" }

Subscribe method:

@mqtt.on_connect()
def connect(client, flags, rc, properties):
    mqtt.client.subscribe("/mqtt") #subscribing mqtt topic
    print("Connected: ", client, flags, rc, properties)

Changing connection params

mqtt_config = MQTTConfig(host = "mqtt.mosquito.org",
    port= 1883,
    keepalive = 60,
    username="username",
    password="strong_password")


mqtt = FastMQTT(
    config=mqtt_config)

✅ Testing

Run tests with pytest, using an external MQTT broker to connect (defaults to 'test.mosquitto.org').

 # (opc) Run a local mosquitto MQTT broker with docker
 $ docker run -d --name mosquitto -p 9001:9001 -p 1883:1883 eclipse-mosquitto:1.6.15
 # Set host for test broker when running pytest
 $ TEST_BROKER_HOST=localhost pytest

Contributing

Fell free to open issue and send pull request.

Thanks To Contributors. Contributions of any kind are welcome!

Before you start please read CONTRIBUTING