fastapi-mqtt

Topics are used by the broker to filter and distribute messages. When designing your topic structure, follow these rules:

Topic Rules

• Case Sensitivity: Topic names are case-sensitive.
• Encoding: Use UTF-8.
• Validation: A topic must consist of at least one character.
• Structure: While topics can start with a /, it is considered bad practice. It is better to use a hierarchical structure without a leading slash (e.g., house/room1/sensor1).

Topic Hierarchy Examples

Valid hierarchical structures include:

• house/room1/sensor1
• house/room2/sensor1
• house-room1-sensor1 (using delimiters other than /)

Publishing Constraints

• No Wildcards: A client can only publish to a specific, individual topic. You cannot use wildcards when publishing. To send the same message to multiple topics, you must publish the message separately for each topic.

FastApi-MQTT provides decorator-based methods to handle specific MQTT lifecycle events and message arrivals. You can implement custom logic by defining functions and decorating them with these hooks:

• on_connect(): Triggered when the client successfully connects to the broker.
• on_disconnect(): Triggered when the client disconnects from the broker.
• on_subscribe(): Triggered when a subscription is successfully established.
• on_message(): Triggered when a new message is received on a subscribed topic.

MQTT settings and configurations are managed using pydantic classes. This allows for structured configuration of:

• Authentication: Providing credentials to authenticate with the MQTT broker.
• Topic Management: Capabilities to unsubscribe from specific topics and publish messages to specific topics.

If you use poetry for dependency management, you can add fastapi-mqtt to your project by entering the poetry shell and using the add command.

poetry shell

poetry add fastapi-mqtt

When running example applications or tests against a local broker (e.g., the Docker container mentioned above), set the TEST_BROKER_HOST environment variable to localhost.

# Run the example app with uvicorn
TEST_BROKER_HOST=localhost uvicorn examples.app:app --port 8000 --reload

# Run the websocket example app
TEST_BROKER_HOST=localhost uvicorn examples.ws_app.app:application --port 8000 --reload

# Run pytest against local broker
TEST_BROKER_HOST=localhost pytest

To use fastapi-mqtt, you need to instantiate an MQTTConfig object for your connection settings and pass it to the FastMQTT client. This client is then integrated with your FastAPI application instance.

from fastapi import FastAPI
from fastapi_mqtt import FastMQTT, MQTTConfig

app = FastAPI()

mqtt_config = MQTTConfig()

mqtt = FastMQTT(
    config=mqtt_config
)

To contribute to the project, you need to set up the development environment using poetry and pre-commit. This includes installing dependencies, activating the virtual environment, and setting up the linting hooks to ensure code quality.

git clone https://github.com/sabuhish/fastapi-mqtt.git
cd fastapi-mqtt
poetry install
# activate the poetry virtualenv
poetry shell
# to make changes and validate them
pre-commit install
pre-commit install-hooks
pre-commit run --all-files
# to run the test suite
pytest

To test your implementation locally, you can run a Mosquitto MQTT broker using Docker with the following command:

docker run -d --name mosquitto -p 9001:9001 -p 1883:1883 eclipse-mosquitto:1.6.15

To ensure the MQTT client starts and shuts down correctly with your FastAPI application, use an asynccontextmanager within the FastAPI lifespan parameter. Call fast_mqtt.mqtt_startup() during startup and fast_mqtt.mqtt_shutdown() during shutdown.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_mqtt import FastMQTT, MQTTConfig

mqtt_config = MQTTConfig()
fast_mqtt = FastMQTT(config=mqtt_config)

@asynccontextmanager
async def _lifespan(_app: FastAPI):
    await fast_mqtt.mqtt_startup()
    yield
    await fast_mqtt.mqtt_shutdown()

app = FastAPI(lifespan=_lifespan)