M5-T10: add paho MQTT client with lifespan connect/reconnect and mqtt/test endpoint
This commit is contained in:
@@ -0,0 +1,257 @@
|
||||
"""MQTT client integration (paho-mqtt 2.x).
|
||||
|
||||
Provides :class:`MqttManager`: a long-lived MQTT client that runs paho's
|
||||
loop in a background thread. Designed to be started in FastAPI's lifespan
|
||||
and shared across the application as a module-level singleton (``mqtt_manager``).
|
||||
|
||||
Key design decisions
|
||||
--------------------
|
||||
- **paho 2.x API**: ``Client`` requires ``CallbackAPIVersion.VERSION2`` as the
|
||||
first positional argument. VERSION2 on_connect callback receives
|
||||
``(client, userdata, connect_flags, reason_code, properties)`` — not the
|
||||
older RC int.
|
||||
- **Background thread**: ``loop_start()`` spawns a daemon thread; ``loop_stop()``
|
||||
joins it cleanly on shutdown.
|
||||
- **Never crash the process**: all paho operations are wrapped in try/except;
|
||||
connection failure is logged but does not propagate to the caller.
|
||||
- **Password safety**: the MQTT password is never passed to ``logger`` calls.
|
||||
- **Reconnect**: ``reconnect(settings)`` tears down the old client and
|
||||
establishes a fresh connection with the new settings. Callers (e.g. PUT
|
||||
/api/config) call this after saving MQTT-related config values.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from app.config import Settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# MQTT settings keys that, when changed, should trigger a reconnect.
|
||||
MQTT_SETTINGS_KEYS = {
|
||||
"mqtt_enabled",
|
||||
"mqtt_broker_host",
|
||||
"mqtt_broker_port",
|
||||
"mqtt_username",
|
||||
"mqtt_password",
|
||||
"mqtt_tls_enabled",
|
||||
}
|
||||
|
||||
|
||||
def _is_configured(settings: Settings) -> bool:
|
||||
"""Return True if MQTT is enabled *and* the broker host is set."""
|
||||
return bool(settings.mqtt_enabled and settings.mqtt_broker_host)
|
||||
|
||||
|
||||
class MqttManager:
|
||||
"""Long-lived MQTT client wrapper.
|
||||
|
||||
Lifecycle
|
||||
---------
|
||||
1. ``connect(settings)`` — start background loop + connect to broker.
|
||||
2. ``publish(...)`` — publish messages while connected.
|
||||
3. ``disconnect()`` — graceful teardown (called on app shutdown).
|
||||
4. ``reconnect(settings)`` — disconnect then re-connect with new settings
|
||||
(called after saving updated MQTT configuration).
|
||||
|
||||
When MQTT is not enabled or not configured, all methods are no-ops.
|
||||
Connection failures are caught and logged; they do not raise.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._client: mqtt.Client | None = None
|
||||
self._lock = threading.Lock()
|
||||
self._connected = False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public properties
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
"""True if the underlying paho client is currently connected."""
|
||||
return self._connected
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public interface
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def is_configured(self, settings: Settings) -> bool:
|
||||
"""Return True if MQTT is enabled and broker host is configured."""
|
||||
return _is_configured(settings)
|
||||
|
||||
def connect(self, settings: Settings) -> None:
|
||||
"""Connect to the MQTT broker and start the background loop thread.
|
||||
|
||||
No-op if MQTT is not enabled or broker host is not set.
|
||||
Connection errors are logged but do not raise.
|
||||
"""
|
||||
if not _is_configured(settings):
|
||||
logger.debug("MQTT not configured or not enabled — skipping connect.")
|
||||
return
|
||||
|
||||
with self._lock:
|
||||
self._start_client(settings)
|
||||
|
||||
def disconnect(self) -> None:
|
||||
"""Disconnect from the broker and stop the background loop thread.
|
||||
|
||||
No-op if no client is active.
|
||||
"""
|
||||
with self._lock:
|
||||
self._stop_client()
|
||||
|
||||
def reconnect(self, settings: Settings) -> None:
|
||||
"""Disconnect the current client (if any) and reconnect with *settings*.
|
||||
|
||||
Call this after saving updated MQTT configuration values.
|
||||
No-op if MQTT is not enabled or broker host is not set.
|
||||
"""
|
||||
with self._lock:
|
||||
self._stop_client()
|
||||
self.connect(settings)
|
||||
|
||||
def publish(
|
||||
self,
|
||||
topic: str,
|
||||
payload: str | bytes | None,
|
||||
*,
|
||||
retain: bool = False,
|
||||
qos: int = 0,
|
||||
) -> None:
|
||||
"""Publish a message to *topic*.
|
||||
|
||||
If the client is not connected the call is silently skipped.
|
||||
Errors are logged but do not raise.
|
||||
"""
|
||||
with self._lock:
|
||||
client = self._client
|
||||
|
||||
if client is None or not self._connected:
|
||||
logger.debug("MQTT publish skipped — not connected (topic=%s).", topic)
|
||||
return
|
||||
|
||||
try:
|
||||
client.publish(topic, payload=payload, qos=qos, retain=retain)
|
||||
except Exception:
|
||||
logger.exception("MQTT publish error (topic=%s).", topic)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers — must be called with self._lock held
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _start_client(self, settings: Settings) -> None:
|
||||
"""Build a fresh paho Client, configure it, and call loop_start + connect."""
|
||||
client = mqtt.Client(
|
||||
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
|
||||
client_id="home-automation",
|
||||
)
|
||||
|
||||
# Callbacks — VERSION2 on_connect signature:
|
||||
# (client, userdata, connect_flags, reason_code, properties)
|
||||
def _on_connect(
|
||||
_client: mqtt.Client,
|
||||
_userdata: object,
|
||||
_flags: mqtt.ConnectFlags,
|
||||
reason_code: mqtt.ReasonCode,
|
||||
_properties: mqtt.Properties | None,
|
||||
) -> None:
|
||||
if reason_code.is_failure:
|
||||
logger.warning("MQTT connection refused: %s", reason_code)
|
||||
self._connected = False
|
||||
else:
|
||||
logger.info("MQTT connected (reason_code=%s).", reason_code)
|
||||
self._connected = True
|
||||
|
||||
# VERSION2 on_disconnect signature:
|
||||
# (client, userdata, disconnect_flags, reason_code, properties)
|
||||
def _on_disconnect(
|
||||
_client: mqtt.Client,
|
||||
_userdata: object,
|
||||
_disconnect_flags: mqtt.DisconnectFlags,
|
||||
reason_code: mqtt.ReasonCode,
|
||||
_properties: mqtt.Properties | None,
|
||||
) -> None:
|
||||
self._connected = False
|
||||
if reason_code.is_failure:
|
||||
logger.warning("MQTT disconnected unexpectedly (reason_code=%s).", reason_code)
|
||||
else:
|
||||
logger.info("MQTT disconnected cleanly.")
|
||||
|
||||
client.on_connect = _on_connect
|
||||
client.on_disconnect = _on_disconnect
|
||||
|
||||
# TLS
|
||||
if settings.mqtt_tls_enabled:
|
||||
try:
|
||||
client.tls_set()
|
||||
except Exception:
|
||||
logger.exception("MQTT TLS setup failed.")
|
||||
return
|
||||
|
||||
# Credentials — never log the password
|
||||
if settings.mqtt_username:
|
||||
client.username_pw_set(
|
||||
username=settings.mqtt_username,
|
||||
password=settings.mqtt_password or None,
|
||||
)
|
||||
|
||||
# Start background loop thread *before* connect so paho can handle
|
||||
# the connection handshake asynchronously.
|
||||
client.loop_start()
|
||||
|
||||
try:
|
||||
client.connect(
|
||||
host=settings.mqtt_broker_host,
|
||||
port=settings.mqtt_broker_port,
|
||||
keepalive=60,
|
||||
)
|
||||
except Exception:
|
||||
# Log without leaking the password
|
||||
logger.exception(
|
||||
"MQTT connect failed (host=%s, port=%s). Stopping loop.",
|
||||
settings.mqtt_broker_host,
|
||||
settings.mqtt_broker_port,
|
||||
)
|
||||
try:
|
||||
client.loop_stop()
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
|
||||
self._client = client
|
||||
logger.info(
|
||||
"MQTT client started (host=%s, port=%s).",
|
||||
settings.mqtt_broker_host,
|
||||
settings.mqtt_broker_port,
|
||||
)
|
||||
|
||||
def _stop_client(self) -> None:
|
||||
"""Disconnect and stop the background loop. Idempotent."""
|
||||
client = self._client
|
||||
if client is None:
|
||||
return
|
||||
self._client = None
|
||||
self._connected = False
|
||||
try:
|
||||
client.disconnect()
|
||||
except Exception:
|
||||
logger.debug("MQTT disconnect raised (ignoring).", exc_info=True)
|
||||
try:
|
||||
client.loop_stop()
|
||||
except Exception:
|
||||
logger.debug("MQTT loop_stop raised (ignoring).", exc_info=True)
|
||||
logger.info("MQTT client stopped.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module-level singleton — shared across lifespan and route handlers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
mqtt_manager = MqttManager()
|
||||
Reference in New Issue
Block a user