fastapi-mqtt is extension for MQTT protocol
Project description
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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for fastapi_mqtt-1.2.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f9de5669b903563fe9ef5b01e818346c78f0916f25dac6da20b6acac0ea6a372 |
|
MD5 | 860fff919c1627659dfeca0ae6d0842c |
|
BLAKE2b-256 | 936c2a9523196c61911d83817f3f5c3adcc371e8dad73834b1967754f6d401aa |