### Install and Setup fastapi-mqtt Project

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/CONTRIBUTING.md

This snippet details the steps to clone the fastapi-mqtt repository, install dependencies using poetry, activate the virtual environment, and set up pre-commit hooks for code validation. It also includes instructions for running the project's test suite.

```shell
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
```

--------------------------------

### Initialize FastMQTT and MQTTConfig in FastAPI

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/docs/getting-started.md

This snippet demonstrates the basic setup of a FastAPI application with the FastMQTT client and MQTTConfig. It shows how to import the necessary classes and instantiate them to prepare for MQTT communication.

```python
from fastapi import FastAPI
from fastapi_mqtt import FastMQTT, MQTTConfig

app = FastAPI()

mqtt_config = MQTTConfig()

mqtt = FastMQTT(
    config=mqtt_config
)

```

--------------------------------

### Run fastapi-mqtt Example Application

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/CONTRIBUTING.md

This command demonstrates how to run the example FastAPI application included in the fastapi-mqtt project using uvicorn. It specifies the application to run, the port, and enables live-reloading for development.

```shell
uvicorn examples.app:app --port 8000 --reload
```

--------------------------------

### Basic FastAPI-MQTT Application Setup and Lifecycle Management

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Sets up a basic FastAPI application with MQTT integration. It configures the MQTT connection, handles connection and disconnection events, subscribes to a topic, and provides an endpoint to publish messages. The setup includes FastAPI's lifespan events for managing MQTT startup and shutdown.

```python
from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI
from gmqtt import Client as MQTTClient
from fastapi_mqtt import FastMQTT, MQTTConfig

# Configure MQTT connection
mqtt_config = MQTTConfig(
    host="mqtt.mosquitto.org",
    port=1883,
    keepalive=60,
    username="your_username",
    password="your_password"
)

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)

@fast_mqtt.on_connect()
def connect(client: MQTTClient, flags: int, rc: int, properties: Any):
    client.subscribe("/mqtt")
    print("Connected: ", client, flags, rc, properties)

@fast_mqtt.on_message()
async def message(client: MQTTClient, topic: str, payload: bytes, qos: int, properties: Any):
    print("Received message: ", topic, payload.decode(), qos, properties)

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

@app.get("/publish")
async def publish_message():
    fast_mqtt.publish("/mqtt", "Hello from FastAPI")
    return {"result": True, "message": "Published"}

```

--------------------------------

### Install fastapi-mqtt

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This command installs the fastapi-mqtt library using pip. It's a prerequisite for using the library in your Python project.

```shell
pip install fastapi-mqtt
```

--------------------------------

### Run WebSocket Example App with Uvicorn

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This command runs a FastAPI application that utilizes WebSocket for MQTT communication with Uvicorn. It connects to a local MQTT broker and enables hot-reloading for development.

```shell
TEST_BROKER_HOST=localhost uvicorn examples.ws_app.app:application --port 8000 --reload
```

--------------------------------

### Topic Subscription with MQTT Wildcards

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Illustrates how to subscribe to MQTT topics using wildcards with FastAPI-MQTT. It shows examples of single-level (+) and multi-level (#) wildcards, as well as setting Quality of Service (QoS) levels for subscriptions, ensuring message delivery guarantees.

```python
from typing import Any
from gmqtt import Client as MQTTClient
from fastapi_mqtt import FastMQTT, MQTTConfig

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

# Subscribe to multiple topics with single-level wildcard (+)
@fast_mqtt.subscribe("mqtt/+/temperature", "mqtt/+/humidity", qos=1)
async def sensor_data(client: MQTTClient, topic: str, payload: bytes, qos: int, properties: Any):
    print(f"Sensor data from {topic}: {payload.decode()}")
    # Example: mqtt/bedroom/temperature -> "22.5"
    # Example: mqtt/kitchen/humidity -> "65"

# Subscribe with multi-level wildcard (#) for all subtopics
@fast_mqtt.subscribe("home/devices/#", qos=0)
async def all_devices(client: MQTTClient, topic: str, payload: bytes, qos: int, properties: Any):
    print(f"Device message: {topic} = {payload.decode()}")
    # Matches: home/devices/living_room/light
    # Matches: home/devices/bedroom/thermostat/status

# High QoS subscription for critical messages
@fast_mqtt.subscribe("alerts/critical/#", qos=2)
async def critical_alerts(client: MQTTClient, topic: str, payload: bytes, qos: int, properties: Any):
    print(f"CRITICAL ALERT: {topic} - {payload.decode()}")
    # QoS 2: Exactly once delivery guaranteed

```

--------------------------------

### Run FastAPI App with Uvicorn

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This command runs the fastapi-mqtt example application using Uvicorn, a high-performance ASGI server. It's configured to connect to a local MQTT broker and reload automatically on code changes.

```shell
# Run the example apps against local broker, with uvicorn
TEST_BROKER_HOST=localhost uvicorn examples.app:app --port 8000 --reload
```

--------------------------------

### Run Pre-commit for Code Formatting

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/CONTRIBUTING.md

This snippet shows how to invoke the pre-commit tool to format all files in the repository before committing changes. It also explains how to install the pre-commit hooks to automatically trigger formatting on commit.

```shell
pre-commit run --all-files
pre-commit install-hooks
```

--------------------------------

### MQTTConfig Configuration Options

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Demonstrates various ways to configure the MQTT connection using the Pydantic-based MQTTConfig model. It covers basic setup, production settings with SSL/TLS, authentication, and advanced SSL context customization.

```python
from ssl import create_default_context
from fastapi_mqtt import MQTTConfig
from gmqtt.mqtt.constants import MQTTv50

# Basic configuration with localhost broker
basic_config = MQTTConfig()

# Production configuration with authentication and reconnection settings
production_config = MQTTConfig(
    host="mqtt.example.com",
    port=8883,
    ssl=True,  # Enable SSL/TLS with default context
    keepalive=120,
    username="production_user",
    password="secure_password",
    version=MQTTv50,
    reconnect_retries=10,
    reconnect_delay=5
)

# Advanced SSL configuration with custom context
ssl_context = create_default_context()
ssl_context.check_hostname = False
ssl_config = MQTTConfig(
    host="secure.mqtt.broker",
    port=8883,
    ssl=ssl_context,
    username="secure_user",
    password="secure_pass"
)

fast_mqtt = FastMQTT(config=production_config)

```

--------------------------------

### FastAPI MQTT Full Integration (Python)

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/docs/example.md

This Python code integrates FastAPI with MQTT using the fastapi-mqtt library. It sets up an MQTT client, defines handlers for connection, message reception, disconnection, and subscription events. It also includes an API endpoint to publish messages to an MQTT topic. Requires fastapi and fastapi-mqtt.

```python
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi_mqtt.config import MQTTConfig
from fastapi_mqtt.fastmqtt import FastMQTT

fast_mqtt = FastMQTT(config=MQTTConfig())

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

app = FastAPI(lifespan=_lifespan)

@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.on_message()
async def message(client, topic, payload, qos, properties):
    print("Received message: ",topic, payload.decode(), qos, properties)
    return 0

@fast_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)

@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():
    fast_mqtt.publish("/mqtt", "Hello from Fastapi") #publishing mqtt topic

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

```

--------------------------------

### Run Tests with pytest and Local Broker

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This command executes pytest tests for fastapi-mqtt. It sets the TEST_BROKER_HOST environment variable to 'localhost', ensuring tests connect to a locally running MQTT broker, likely started with Docker.

```shell
# Set host for test broker when running pytest
TEST_BROKER_HOST=localhost pytest
```

--------------------------------

### FastAPI-MQTT: Legacy Initialization Method

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Illustrates the legacy method of initializing FastMQTT using `init_app` for older FastAPI versions, before the adoption of lifespan events. It sets up event handlers like `@fast_mqtt.on_connect()` for managing MQTT client lifecycle events and demonstrates publishing a message from a FastAPI endpoint.

```python
from fastapi import FastAPI
from fastapi_mqtt import FastMQTT, MQTTConfig

mqtt_config = MQTTConfig(host="mqtt.broker.com", port=1883)
fast_mqtt = FastMQTT(config=mqtt_config)

app = FastAPI()

# Legacy initialization method (deprecated in favor of lifespan)
fast_mqtt.init_app(app)

@fast_mqtt.on_connect()
def connect(client, flags, rc, properties):
    client.subscribe("/test/topic")
    print("Connected via legacy init_app method")

@app.get("/test")
async def test_endpoint():
    fast_mqtt.publish("/test/topic", "Test message")
    return {"result": "published"}
```

--------------------------------

### Initialize FastMQTT with Custom Client Configuration

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Configure the FastMQTT client with custom settings such as client ID, session persistence, and optimistic acknowledgements. Demonstrates initializing FastMQTT with specific MQTT broker details.

```python
import uuid
from fastapi_mqtt import FastMQTT, MQTTConfig

# Default configuration with auto-generated client ID
mqtt_config = MQTTConfig(host="mqtt.broker.com")
fast_mqtt_auto = FastMQTT(config=mqtt_config)

# Example with custom client ID and other settings (not fully shown)
# mqtt_config_custom = MQTTConfig(
#     host="mqtt.broker.com",
#     client_id=str(uuid.uuid4()),
#     keepalive=60,
#     will_message_topic="/status/app",
#     will_message_payload="App disconnected",
#     will_qos=1,
#     will_retain=False,
#     auth={ "username": "user", "password": "password" },
#     tls_enabled=False,
#     tls_ca_certs=None,
#     tls_certfile=None,
#     tls_keyfile=None,
#     tls_insecure=False
# )
# fast_mqtt_custom = FastMQTT(config=mqtt_config_custom)
```

--------------------------------

### Basic FastAPI App with MQTT Integration

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This Python code demonstrates a basic FastAPI application integrated with fastapi-mqtt. It shows how to set up the MQTT client, define connection and message handling callbacks, and publish messages. The application uses an async context manager for MQTT startup and shutdown.

```python
from contextlib import asynccontextmanager
from typing import Any

from fastapi import FastAPI
from gmqtt import Client as MQTTClient

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)


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

@fast_mqtt.subscribe("mqtt/+/temperature", "mqtt/+/humidity", qos=1)
async def home_message(client: MQTTClient, topic: str, payload: bytes, qos: int, properties: Any):
    print("temperature/humidity: ", topic, payload.decode(), qos, properties)

@fast_mqtt.on_message()
async def message(client: MQTTClient, topic: str, payload: bytes, qos: int, properties: Any):
    print("Received message: ", topic, payload.decode(), qos, properties)

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

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

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

@app.get("/test")
async def func():
    fast_mqtt.publish("/mqtt", "Hello from Fastapi")  # publishing mqtt topic
    return {"result": True, "message": "Published"}

```

--------------------------------

### Run Local MQTT Broker with Docker

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This command uses Docker to run a local Mosquitto MQTT broker. It exposes the necessary ports (9001 for WebSockets, 1883 for MQTT) and is useful for local testing.

```shell
# (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
```

--------------------------------

### Handle MQTT Connection Events with FastAPI

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Implement lifecycle event handlers for MQTT connection, disconnection, and subscription confirmation in FastAPI. Utilizes gmqtt.Client and fastapi-mqtt.

```python
from typing import Any
from gmqtt import Client as MQTTClient
from fastapi_mqtt import FastMQTT, MQTTConfig

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

@fast_mqtt.on_connect()
def handle_connect(client: MQTTClient, flags: int, rc: int, properties: Any):
    # Subscribe to topics on successful connection
    client.subscribe("/mqtt/commands")
    client.subscribe("sensors/#")
    print(f"Connected with result code: {rc}")
    print(f"Session present: {flags & 0x01}")

@fast_mqtt.on_disconnect()
def handle_disconnect(client: MQTTClient, packet, exc=None):
    if exc:
        print(f"Unexpected disconnection: {exc}")
    else:
        print("Clean disconnection")

@fast_mqtt.on_subscribe()
def handle_subscribe(client: MQTTClient, mid: int, qos: int, properties: Any):
    print(f"Subscribed with message ID: {mid}, QoS: {qos}")
```

--------------------------------

### Unsubscribe and Subscribe to MQTT Topics Dynamically with FastAPI

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Dynamically manage MQTT subscriptions by unsubscribing from and subscribing to topics at runtime within a FastAPI application. Uses fastapi-mqtt for topic management.

```python
from fastapi import FastAPI
from fastapi_mqtt import FastMQTT, MQTTConfig

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

# Lifespan setup omitted for brevity
app = FastAPI()

@fast_mqtt.on_connect()
def connect(client, flags, rc, properties):
    client.subscribe("sensors/+/temperature")
    client.subscribe("sensors/+/humidity")
    print("Subscribed to all sensors")

@app.delete("/subscribe/{sensor_type}")
async def unsubscribe_sensor(sensor_type: str):
    topic = f"sensors/+/{sensor_type}"
    fast_mqtt.unsubscribe(topic)
    return {"status": "unsubscribed", "topic": topic}

@app.post("/subscribe/{sensor_type}")
async def subscribe_sensor(sensor_type: str):
    topic = f"sensors/+/{sensor_type}"
    fast_mqtt.client.subscribe(topic)
    return {"status": "subscribed", "topic": topic}
```

--------------------------------

### Subscribe to MQTT Topic on Connect

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This Python function is an MQTT connection callback. When the client successfully connects to the MQTT broker, it automatically subscribes to the '/mqtt' topic.

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

--------------------------------

### FastAPI-MQTT: Custom Client ID and Persistent Session Configuration

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Configures FastMQTT instances with custom client IDs and demonstrates the difference between persistent (clean_session=False) and non-persistent (clean_session=True) sessions. Persistent sessions maintain subscriptions across reconnections, while clean sessions do not. The optimistic_acknowledgement parameter controls whether to wait for broker acknowledgment.

```python
import uuid
from fastapi_mqtt import FastMQTT, MQTTConfig

mqtt_config = MQTTConfig(host="mqtt.broker.com", port=1883)

# Custom client ID with persistent session
custom_client_id = f"fastapi-app-{uuid.uuid4().hex[:8]}"
fast_mqtt_persistent = FastMQTT(
    config=mqtt_config,
    client_id=custom_client_id,
    clean_session=False,  # Maintain session across reconnections
    optimistic_acknowledgement=True  # Don't wait for broker ACK
)

# Clean session (no persistence)
fast_mqtt_clean = FastMQTT(
    config=mqtt_config,
    client_id="temporary-client",
    clean_session=True,  # Don't restore subscriptions on reconnect
    optimistic_acknowledgement=False  # Wait for broker ACK
)

print(f"Client ID: {fast_mqtt_persistent.client._client_id}")
print(f"Clean session: {fast_mqtt_persistent.client._clean_session}")
```

--------------------------------

### Configure MQTT Connection Parameters

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This Python code snippet shows how to configure custom connection parameters for the fastapi-mqtt client. It allows specifying the MQTT broker host, port, keepalive interval, and authentication credentials.

```python
mqtt_config = MQTTConfig(
    host="mqtt.mosquito.org",
    port=1883,
    keepalive=60,
    username="username",
    password="strong_password",
)
fast_mqtt = FastMQTT(config=mqtt_config)
```

--------------------------------

### Publish MQTT Messages with FastAPI

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Publish messages to MQTT topics from FastAPI endpoints or background tasks. Supports various QoS levels (0, 1, 2) and retain flags. Requires fastapi and fastapi-mqtt.

```python
from fastapi import FastAPI, BackgroundTasks
from fastapi_mqtt import FastMQTT, MQTTConfig

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

# Lifespan setup omitted for brevity
app = FastAPI()

@app.post("/sensor/temperature")
async def publish_temperature(value: float, location: str):
    # Basic publish with QoS 0 (at most once)
    fast_mqtt.publish(f"sensors/{location}/temperature", str(value), qos=0)
    return {"status": "published", "topic": f"sensors/{location}/temperature"}

@app.post("/alert")
async def publish_alert(message: str):
    # Publish with QoS 1 (at least once) and retain flag
    fast_mqtt.publish(
        "alerts/system",
        message,
        qos=1,
        retain=True  # Broker retains message for new subscribers
    )
    return {"status": "alert_sent", "message": message}

@app.post("/status")
async def publish_device_status(device_id: str, status: str, background_tasks: BackgroundTasks):
    # Publish in background task
    def send_status():
        fast_mqtt.publish(
            f"devices/{device_id}/status",
            status,
            qos=2  # Exactly once delivery
        )

    background_tasks.add_task(send_status)
    return {"device": device_id, "status": status}
```

--------------------------------

### Publish MQTT Message

Source: https://github.com/sabuhish/fastapi-mqtt/blob/master/README.md

This Python function demonstrates how to publish a message to an MQTT topic using the fastapi-mqtt client. It's designed to be called within a FastAPI application context.

```python
async def func():
    fast_mqtt.publish("/mqtt", "Hello from Fastapi")  # publishing mqtt topic
    return {"result": True, "message": "Published"}
```

--------------------------------

### Configure MQTT Will Message in FastAPI

Source: https://context7.com/sabuhish/fastapi-mqtt/llms.txt

Set up Last Will and Testament messages for abnormal client disconnections using FastAPI and fastapi-mqtt. Configures the broker to send a message on behalf of the client if it disconnects unexpectedly.

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

# Configure will message for abnormal disconnection
mqtt_config = MQTTConfig(
    host="mqtt.broker.com",
    port=1883,
    will_message_topic="/status/app",
    will_message_payload="Application disconnected unexpectedly",
    will_delay_interval=5  # Delay in seconds before sending will message
)

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)

@fast_mqtt.on_connect()
def connect(client, flags, rc, properties):
    # Subscribe to status topic to monitor will messages
    client.subscribe("/status/#")
    print("Connected - will message configured")

@fast_mqtt.on_message()
async def message(client, topic, payload, qos, properties):
    if topic.startswith("/status/"):
        print(f"Status update: {payload.decode()}")
    return 0

@app.get("/health")
async def health_check():
    return {"status": "healthy", "mqtt": "connected"}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.