Project: a small Python MQTT-to-REST bridge with SQLite history, consumed by Home Assistant

[ZerOne]

Goal

Build a small service that listens to temperature readings on MQTT, keeps the latest value per room plus a history in SQLite, and serves the latest values as read-only JSON over HTTP. Then have Home Assistant read it with its RESTful integration. It's a practical way to practise MQTT, a tiny web API and some SQL in one go. It uses only the Python standard library plus paho-mqtt. I ran it against a local Mosquitto broker and checked the endpoints with curl, but I haven't run it long term, so treat it as a starting point. The MQTT basics are in the MQTT starter thread I posted in this forum.

Assumed message format: topic home/<room>/temperature with a JSON payload like {"c": 22.4}. Adjust the parsing for your own devices.

Step 1: Setup

Code: Select all

python3 -m venv venv
./venv/bin/pip install paho-mqtt

I used paho-mqtt 2.1.0. For credentials, set MQTT_USER and MQTT_PASSWORD in the environment rather than in the file.

Step 2: bridge.py

Code: Select all

import json, os, sqlite3, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "127.0.0.1")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USER, PASSWORD = os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASSWORD")
HTTP_BIND = os.environ.get("HTTP_BIND", "127.0.0.1")
HTTP_PORT = int(os.environ.get("HTTP_PORT", "8099"))

lock = threading.Lock()
db = sqlite3.connect("readings.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS readings (ts REAL, room TEXT, celsius REAL)")
latest = {}

def on_connect(client, userdata, flags, reason_code, properties):
    if not reason_code.is_failure:
        client.subscribe("home/+/temperature", qos=1)

def on_message(client, userdata, msg):
    try:
        room = msg.topic.split("/")[1]
        celsius = float(json.loads(msg.payload)["c"])
    except (ValueError, KeyError, TypeError, IndexError):
        return  # ignore anything malformed
    now = time.time()
    with lock:
        latest[room] = {"celsius": celsius, "ts": now}
        db.execute("INSERT INTO readings VALUES (?, ?, ?)", (now, room, celsius))
        db.commit()

class Handler(BaseHTTPRequestHandler):
    def _send(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        parts = [p for p in self.path.split("?")[0].split("/") if p]
        with lock:
            if parts == ["health"]:
                return self._send(200, {"status": "ok"})
            if parts == ["latest"]:
                return self._send(200, dict(latest))
            if len(parts) == 2 and parts[0] == "latest":
                if parts[1] in latest:
                    return self._send(200, latest[parts[1]])
                return self._send(404, {"error": "unknown room"})
        self._send(404, {"error": "not found"})

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="mqtt-rest-bridge")
if USER:
    client.username_pw_set(USER, PASSWORD)
client.on_connect, client.on_message = on_connect, on_message
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()
try:
    ThreadingHTTPServer((HTTP_BIND, HTTP_PORT), Handler).serve_forever()
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Step 3: Test it

Code: Select all

./venv/bin/python bridge.py &
mosquitto_pub -t home/lounge/temperature -m '{"c": 22.4}'
curl http://127.0.0.1:8099/latest
curl http://127.0.0.1:8099/latest/lounge
curl -i http://127.0.0.1:8099/latest/nope   # expect 404

Step 4: Read it from Home Assistant

Home Assistant's RESTful integration polls an endpoint and pulls values out with a template. Set HTTP_BIND to the LAN address of the machine running the bridge (not 0.0.0.0 unless you mean it), then add this to configuration.yaml and restart:

Code: Select all

rest:
  - resource: http://192.168.1.50:8099/latest/lounge
    scan_interval: 60
    sensor:
      - name: "Lounge bridge temperature"
        value_template: "{{ value_json.celsius }}"
        unit_of_measurement: "°C"
        device_class: temperature

The docs say the default timeout is 10 seconds, and that the sensor becomes unavailable if it is reached. Note the 404 for a room that has not reported yet, so the sensor will be unavailable until the first reading arrives.

Ideas to extend it

• Add a /history/<room> endpoint that reads from SQLite (use parameterised queries, as above, never string formatting).
• Prune old rows on a timer.
• Rewrite the API with a framework once it grows, or run it under systemd so it restarts.

Keep it safe: this API has no authentication, so keep it on a trusted LAN or localhost and don't port-forward it. Don't put broker passwords in the script. Don't build SQL from message contents.

Discussion: How would you add authentication to something like this, and would you rather use a framework such as Flask or FastAPI for the HTTP side?