fastapi-mqtt is extension for MQTT protocol

Last update: Dec 28, 2022

Overview

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_names)
Base Settings available with pydantic class
Authetication 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
)

@app.on_event("startup")
async def startapp():
    await mqtt.connection()

@app.on_event("shutdown")
async def shutdown():
    await mqtt.client.disconnect()

@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.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():
    await 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)

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

Comments

Multiple subscriptions with multiple workers

Hello,

I just recently switch to fastpi-mqtt for some testing. While developing with a single worker everything works fine as soon as I put the code in a docker instance I noticed that it has issues with multiple workers though. Per default tiangolo/uvicorn-gunicorn-fastapi creates 8 worker processes for me, as a result I have 8 subscriptions as well and if a change is published my on_message method is called 8 times which is not ideal. Either I am missing something or there are issues with multiple workers.

opened by maru-sama 8
QOS != 0 gives exception
Hello,

When publishing with QOS of 1 or 2 I get the following stack trace

Traceback (most recent call last): File "e:\code\python\pythontest\venv\lib\site-packages\uvicorn\protocols\http\h11_impl.py", line 396, in run_asgi result = await app(self.scope, self.receive, self.send) File "e:\code\python\pythontest\venv\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 45, in call return await self.app(scope, receive, send) File "e:\code\python\pythontest\venv\lib\site-packages\fastapi\applications.py", line 199, in call await super().call(scope, receive, send) File "e:\code\python\pythontest\venv\lib\site-packages\starlette\applications.py", line 111, in call await self.middleware_stack(scope, receive, send) File "e:\code\python\pythontest\venv\lib\site-packages\starlette\middleware\errors.py", line 181, in call raise exc from None File "e:\code\python\pythontest\venv\lib\site-packages\starlette\middleware\errors.py", line 159, in call await self.app(scope, receive, _send) File "e:\code\python\pythontest\venv\lib\site-packages\starlette\exceptions.py", line 82, in call raise exc from None File "e:\code\python\pythontest\venv\lib\site-packages\starlette\exceptions.py", line 71, in call await self.app(scope, receive, sender) File "e:\code\python\pythontest\venv\lib\site-packages\starlette\routing.py", line 566, in call await route.handle(scope, receive, send) File "e:\code\python\pythontest\venv\lib\site-packages\starlette\routing.py", line 227, in handle await self.app(scope, receive, send) File "e:\code\python\pythontest\venv\lib\site-packages\starlette\routing.py", line 41, in app response = await func(request) File "e:\code\python\pythontest\venv\lib\site-packages\fastapi\routing.py", line 201, in app raw_response = await run_endpoint_function( File "e:\code\python\pythontest\venv\lib\site-packages\fastapi\routing.py", line 148, in run_endpoint_function return await dependant.call(**values) File ".\main.py", line 60, in func await fast_mqtt.publish("test", "Hello from Fastapi", qos=1) #publishing mqtt topic File "e:\code\python\pythontest\venv\lib\site-packages\fastapi_mqtt\fastmqtt.py", line 226, in publish return await self.loop.run_in_executor(self.executor, func) File "C:\Python38\lib\concurrent\futures\thread.py", line 57, in run result = self.fn(*self.args, **self.kwargs) File "e:\code\python\pythontest\venv\lib\site-packages\gmqtt\client.py", line 291, in publish self._persistent_storage.push_message_nowait(mid, package) File "e:\code\python\pythontest\venv\lib\site-packages\gmqtt\storage.py", line 12, in push_message_nowait return asyncio.ensure_future(self.push_message(mid, raw_package)) File "C:\Python38\lib\asyncio\tasks.py", line 660, in ensure_future loop = events.get_event_loop() File "C:\Python38\lib\asyncio\events.py", line 639, in get_event_loop raise RuntimeError('There is no current event loop in thread %r.' RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'. C:\Python38\lib\asyncio\base_events.py:1860: RuntimeWarning: coroutine 'HeapPersistentStorage.push_message' was never awaited handle = None # Needed to break cycles when an exception occurs.

Using QOS of 0 works fine

I took the example on the site and change the config to my server and managed to reproduce it. Here it is:

from fastapi_mqtt.fastmqtt import FastMQTT from fastapi import FastAPI from fastapi_mqtt.config import MQQTConfig import ssl

mqtt_config = MQQTConfig()

def config(): server_cert = client_cert = client_key =

context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=server_cert) context.load_cert_chain(certfile=client_cert, keyfile=client_key) mqtt_config.ssl = context mqtt_config.password = <MY PASSWORD> mqtt_config.username = <MY USERNAME> mqtt_config.host = <MY SERVER> mqtt_config.port = 8883

config()

fast_mqtt = FastMQTT( config=mqtt_config )

app = FastAPI()

fast_mqtt.init_app(app)

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

@fast_mqtt.subscribe("mqtt/+/temperature", "mqtt/+/humidity") async def home_message(client, topic, payload, qos, properties): print("temperature/humidity: ", topic, payload.decode(), qos, properties) return 0

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

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

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

@app.get("/") async def func(): await fast_mqtt.publish("test", "Hello from Fastapi", qos=1) #publishing mqtt topic

return {"result": True,"message":"Published" }
opened by alanaquilina 8
got Future attached to a different loop
When I run the app using uvicorn directly, everything is fine

However when I run it using an external file, this raises the error

got Future attached to a different loop

I made a repository to describe my situation, could you give it a look. https://github.com/ngochieu642/issue-3-fastapi-mqtt
opened by ngochieu642 6
Example code not working

Hi, I tried running the example code and got the following error: OSError: Multiple exceptions: [Errno 61] Connect call failed ('::1', 1883, 0, 0), [Errno 61] Connect call failed ('127.0.0.1', 1883) I'd like to convert my http API to a mqtt for better IoT support but this happens. rant it with uvicorn main:app --reload command thx in advance

Running with hyper corn gives this error: ../fastapi_mqtt/fastmqtt.py", line 78, in init log_info = logger NameError: name 'logger' is not defined

opened by jruokolainen 4
Insert MQTT Messages into Database

As requested by @sabuhish I have added this here as an issue. Here's the link to my original post.

TL;DR: I was not able to implement the insertion of MQTT messages received by fastapi-mqtt into my database. I figured a work-around using paho-mqtt in another python script that makes a post request to the API, but of course, it's not nearly as graceful as it would be if fastapi could do this all by itself. i.e. Connecting -> Subscribing -> Receiving -> Insert to DB

P.S. : Thanks fastapi-mqtt devs for making this library. Your work is witnessed and appreciated!!!

opened by wiserdivisor 4

Add an ability to move mqtt handlers outside main

Hi @sabuhish, thanks for the great library!

I'm wondering if it's possible to move mqtt handlers to a different package rather than putting everything into main?

Usually, when you build a production-level code, you organize it into logical packages, e.g. api, model, etc. FastAPI has a good concept of APIRouter as an example.

If I run a sample from the README within a script where FastAPI instance is created, everything works as expected. But when I try to move mqtt code somewhere else (outside main), the app is failing at the start:

ERROR:    Traceback (most recent call last):
  File "/project_path/.venv/lib/python3.8/site-packages/starlette/routing.py", line 621, in lifespan
    async with self.lifespan_context(app):
  File "/project_path/.venv/lib/python3.8/site-packages/starlette/routing.py", line 518, in __aenter__
    await self._router.startup()
  File "/project_path/.venv/lib/python3.8/site-packages/starlette/routing.py", line 598, in startup
    await handler()
  File "/project_path/app/main.py", line 77, in startup
    await mqtt_client.mqtt.connection()
  File "/project_path/.venv/lib/python3.8/site-packages/fastapi_mqtt/fastmqtt.py", line 129, in connection
    await self.client.connect(self.client._host,self.client._port,self.client._ssl,self.client._keepalive,version)
  File "/project_path/.venv/lib/python3.8/site-packages/gmqtt/client.py", line 230, in connect
    await self._connected.wait()
  File "/usr/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
RuntimeError: Task <Task pending name='Task-3' coro=<LifespanOn.main() running at /project_path/.venv/lib/python3.8/site-packages/uvicorn/lifespan/on.py:84>> got Future <Future pending> attached to a different loop

As far as I understood, mqtt instance is running in a different event loop for some reason if it's located in the other package? Is there any workaround for that?

P.S. I run the code via uvicorn.

opened by sskorol 3

Pass MQTT client to a router
Hi,

I would like to know if it is possible to pass the MQTT client initiated by init_app() to a router that is included by included_router()?

mqtt_config = MQTTConfig() mqtt = FastMQTT( config=mqtt_config ) app = FastAPI() mqtt.init_app(app) app.include_router(activities.router)

I would like to access the @mqtt.on_message() decorator from the activities.router routes to read the message only when a specific route is called.

Thanks for your help.
opened by goldyfruit 2
Is this project dead?

There seem to be issues even running sample code - likely some changes in gunicorn. They've been outlined in other issues Is this project still being worked on?

opened by zabrewer 2
Add documentation for subscribe decorator

Hello,

I really like your library, but I really missed the ability for topic specific handlers. I could only find that this feature was already implemented by digging into the source code. It is only used in the examples at a single point, but never mentioned in the readme or documentation.

Therefore I added documentation for the subscribe decorator to the readme and website. Hopefully this helps other users to find this feature more quickly.

Thanks for your work,

Jannik

opened by spontanurlaub 1

Suggestion: Update requirement to newest fastapi automatically

PS E:\xx\libs> pip install -U fastapi-mqtt --proxy=http://localhost:6666
Collecting fastapi-mqtt
  Downloading fastapi_mqtt-1.0.4-py3-none-any.whl (8.0 kB)
Collecting fastapi<0.76.0,>=0.75.1
  Using cached fastapi-0.75.2-py3-none-any.whl (54 kB)
Collecting uvicorn<0.18.0,>=0.17.6
  Downloading uvicorn-0.17.6-py3-none-any.whl (53 kB)
     ---------------------------------------- 53.6/53.6 kB 173.4 kB/s eta 0:00:00
Requirement already satisfied: gmqtt<0.7.0,>=0.6.11 in c:\python310\lib\site-packages (from fastapi-mqtt) (0.6.11)
Requirement already satisfied: pydantic<2.0.0,>=1.9.0 in c:\python310\lib\site-packages (from fastapi-mqtt) (1.9.0)
Collecting starlette==0.17.1
  Using cached starlette-0.17.1-py3-none-any.whl (58 kB)
Requirement already satisfied: anyio<4,>=3.0.0 in c:\python310\lib\site-packages (from starlette==0.17.1->fastapi<0.76.0,>=0.75.1->fastapi-mqtt) (3.5.0)
Requirement already satisfied: typing-extensions>=3.7.4.3 in c:\python310\lib\site-packages (from pydantic<2.0.0,>=1.9.0->fastapi-mqtt) (3.10.0.0)
Requirement already satisfied: click>=7.0 in c:\python310\lib\site-packages (from uvicorn<0.18.0,>=0.17.6->fastapi-mqtt) (8.0.4)
Requirement already satisfied: asgiref>=3.4.0 in c:\python310\lib\site-packages (from uvicorn<0.18.0,>=0.17.6->fastapi-mqtt) (3.5.0)
Requirement already satisfied: h11>=0.8 in c:\python310\lib\site-packages (from uvicorn<0.18.0,>=0.17.6->fastapi-mqtt) (0.12.0)
Requirement already satisfied: colorama in c:\python310\lib\site-packages (from click>=7.0->uvicorn<0.18.0,>=0.17.6->fastapi-mqtt) (0.4.4)
Requirement already satisfied: sniffio>=1.1 in c:\python310\lib\site-packages (from anyio<4,>=3.0.0->starlette==0.17.1->fastapi<0.76.0,>=0.75.1->fastapi-mqtt) (1.2.0)
Requirement already satisfied: idna>=2.8 in c:\python310\lib\site-packages (from anyio<4,>=3.0.0->starlette==0.17.1->fastapi<0.76.0,>=0.75.1->fastapi-mqtt) (3.2)
Installing collected packages: uvicorn, starlette, fastapi, fastapi-mqtt
  Attempting uninstall: uvicorn
    Found existing installation: uvicorn 0.15.0
    Uninstalling uvicorn-0.15.0:
      Successfully uninstalled uvicorn-0.15.0
  Attempting uninstall: starlette
    Found existing installation: starlette 0.19.1
    Uninstalling starlette-0.19.1:
      Successfully uninstalled starlette-0.19.1
  Attempting uninstall: fastapi
    Found existing installation: fastapi 0.77.1
    Uninstalling fastapi-0.77.1:
      Successfully uninstalled fastapi-0.77.1
Successfully installed fastapi-0.75.2 fastapi-mqtt-1.0.4 starlette-0.17.1 uvicorn-0.17.6

opened by honglei 1

Changed logging from info to debug and fixed typos

It seems like it would be better behavior to have the users specify their own logging messages rather than hitting them with the debug logs. I changed the current logs to the debug level because it creates a cleaner experience when looking at the output.

opened by petercinibulk 1
Fix dependencies uvicorn blue

Problem

Unvicorn is fixed to "^0.19.0", which translates to ">=0.19.0, <=0.20.0". However, there is no direct dependency on unvicorn except the logger that is imported.

Solution

This PR resolves this issue by relaxing the dependency. Also, some type annotation and linter errors are fixed.

(Functionality stays unchanged)

opened by mblo 0
Error run fastapi-mqtt using gunicorn

I use fastapi-mqtt==0.3.0, python==3.9.7, fastapi==0.68.1 I run example application by command : uvicorn app:app --host 0.0.0.0 --port 7000 --reload Everything is ok -> app run

But: I run example application by command : gunicorn app:app -k uvicorn.workers.UvicornWorker --workers=9 -b 0.0.0.0:7000

Error : File "/home/hoanganh/smarthome-server/venv/lib/python3.9/site-packages/gmqtt/client.py", line 230, in connect await self._connected.wait() File "/usr/lib/python3.9/asyncio/locks.py", line 226, in wait await fut RuntimeError: Task <Task pending name='Task-3' coro=<LifespanOn.main() running at /home/hoanganh/smarthome-server/venv/lib/python3.9/site-packages/uvicorn/lifespan/on.py:84>> got Future attached to a different loop

[2021-09-21 20:42:13 +0700] [10419] [ERROR] Application startup failed. Exiting. [2021-09-21 20:42:13 +0700] [10419] [INFO] Worker exiting (pid: 10419) Task was destroyed but it is pending! task: <Task pending name='Task-1' coro=<Client._resend_qos_messages() running at /home/hoanganh/smarthome-server/venv/lib/python3.9/site-packages/gmqtt/client.py:176>> sys:1: RuntimeWarning: coroutine 'Client._resend_qos_messages' was never awaited [2021-09-21 20:42:14 +0700] [10418] [INFO] Shutting down: Master [2021-09-21 20:42:14 +0700] [10418] [INFO] Reason: Worker failed to boot.

opened by ruem2802 3

error Application startup failed. Exiting. when running!

Hi there. I implemented codes example but I get this error when uvicorn runs:

  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/starlette/routing.py", line 540, in lifespan
    async for item in self.lifespan_context(app):
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/starlette/routing.py", line 481, in default_lifespan
    await self.startup()
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/starlette/routing.py", line 516, in startup
    await handler()
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/fastapi_mqtt/fastmqtt.py", line 276, in startup
    await self.connection()
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/fastapi_mqtt/fastmqtt.py", line 128, in connection
    await self.client.connect(self.client._host,self.client._port,self.client._ssl,self.client._keepalive,version)
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/gmqtt/client.py", line 225, in connect
    self._connection = await self._create_connection(
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/gmqtt/client.py", line 241, in _create_connection
    connection = await MQTTConnection.create_connection(host, port, ssl, clean_session, keepalive)
  File "/home/ali/Documents/first_fastapi/venv/lib/python3.9/site-packages/gmqtt/mqtt/connection.py", line 27, in create_connection
    transport, protocol = await loop.create_connection(MQTTProtocol, host, port, ssl=ssl)
  File "uvloop/loop.pyx", line 2024, in create_connection
  File "uvloop/loop.pyx", line 2001, in uvloop.loop.Loop.create_connection
ConnectionRefusedError: [Errno 111] Connection refused

ERROR:    Application startup failed. Exiting.

opened by hossainirad 8

Releases(1.0.6)

1.0.6(Nov 24, 2022)

Source code(tar.gz)
Source code(zip)
1.0.5(Jun 9, 2022)

Source code(tar.gz)
Source code(zip)
1.0.4(Apr 19, 2022)

This release adds typing information to the callback functions. #33

Thanks to @mblo
Source code(tar.gz)
Source code(zip)
1.0.3(Apr 17, 2022)

Source code(tar.gz)
Source code(zip)
1.0.2(Apr 17, 2022)

updated pypi documentation
Source code(tar.gz)
Source code(zip)
1.0.1(Apr 17, 2022)

Added deployment with poetry
Source code(tar.gz)
Source code(zip)
1.0.0(Apr 17, 2022)
This release initializes version 1.0.0 with major changes, some can be breaking. In this release I have covered the followings:

Refactored codebase, created separate handler file

Updated documentation

Changed deployment flow to PyPI

Added poetry as dependency management instead of setuptools

Added Makefile to use of make, having said that we now use mypy, isort, blue, flake8 libraries for development

Brought class MQTTHandlers

Source code(tar.gz)
Source code(zip)
0.4.0(Oct 9, 2021)

Highlighted changes and updates on this release. v0.4.0

The library included an update of static type annotations. Changed ingo log level to debug.

@mblo and @petercinibulk Thank you for taking the time and giving effort.
Source code(tar.gz)
Source code(zip)
v0.3.0(Apr 5, 2021)

Source code(tar.gz)
Source code(zip)
v0.2.0(Feb 11, 2021)

Source code(tar.gz)
Source code(zip)
0.1.1(Jan 1, 2021)

Source code(tar.gz)
Source code(zip)
0.1.0(Dec 5, 2020)

Source code(tar.gz)
Source code(zip)
0.0.5(Nov 22, 2020)

Source code(tar.gz)
Source code(zip)
0.0.4(Nov 22, 2020)

Source code(tar.gz)
Source code(zip)
0.0.3(Nov 21, 2020)

Source code(tar.gz)
Source code(zip)
0.0.2(Nov 21, 2020)

Source code(tar.gz)
Source code(zip)
0.0.1(Nov 21, 2020)

Source code(tar.gz)
Source code(zip)

Owner

Sabuhi

open science enthusiast loves python, go, and shell. Enjoys Linux environment.

GitHub Repository https://sabuhish.github.io/fastapi-mqtt/

fastapi-mqtt is extension for MQTT protocol

Related tags

Overview

fastapi-mqtt

Documentation: FastApi-MQTT

🔨 Installation

🕹 Guide

Contributing

Comments

Releases(1.0.6)

1.0.6(Nov 24, 2022)

1.0.5(Jun 9, 2022)

1.0.4(Apr 19, 2022)

1.0.3(Apr 17, 2022)

1.0.2(Apr 17, 2022)

1.0.1(Apr 17, 2022)

1.0.0(Apr 17, 2022)

0.4.0(Oct 9, 2021)

v0.3.0(Apr 5, 2021)

v0.2.0(Feb 11, 2021)

0.1.1(Jan 1, 2021)

0.1.0(Dec 5, 2020)

0.0.5(Nov 22, 2020)

0.0.4(Nov 22, 2020)

0.0.3(Nov 21, 2020)

0.0.2(Nov 21, 2020)

0.0.1(Nov 21, 2020)

Owner

Sabuhi

Hyperlinks for pydantic models

Deploy/View images to database sqlite with fastapi

Fastapi-ml-template - Fastapi ml template with python

Dead-simple mailer micro-service for static websites

Cookiecutter API for creating Custom Skills for Azure Search using Python and Docker

스타트업 개발자 채용

Slack webhooks API served by FastAPI

A kedro-plugin to serve Kedro Pipelines as API

Keepalive - Discord Bot to keep threads from expiring

Voucher FastAPI

Twitter API monitor with fastAPI + MongoDB

fastapi-mqtt is extension for MQTT protocol

🐞 A debug toolbar for FastAPI based on the original django-debug-toolbar. 🐞

A set of demo of deploying a Machine Learning Model in production using various methods

Stac-fastapi built on Tile38 and Redis to support caching

Hook Slinger acts as a simple service that lets you send, retry, and manage event-triggered POST requests, aka webhooks

Flask + marshmallow for beautiful APIs

LuSyringe is a documentation injection tool for your classes when using Fast API

A Python framework to build Slack apps in a flash with the latest platform features.

cookiecutter template for web API with python