Envoyer des données de MQTT vers SQLite avec Python est le moyen le plus simple de doter votre projet IoT d'un historique : un seul fichier, aucun serveur de base de données et tout le SQL pour vos requêtes. Dans ce tutoriel, vous allez construire un subscriber paho-mqtt qui stocke chaque message de broker.connectmq.com avec son horodatage, son topic et ses flags, écrit par lots efficaces et survit aux reconnexions. Vous interrogerez ensuite les données pour obtenir les valeurs récentes, des moyennes horaires et faire le ménage selon une durée de rétention.
Pourquoi SQLite pour les données MQTT ?
Un broker MQTT livre les messages à ceux qui sont abonnés à l'instant présent. Ce n'est pas une base de données et, en dehors d'un retained message par topic, il ne conserve pas d'historique. Si vous voulez répondre à « quelle était la température à 3 heures du matin ? », quelque chose doit s'abonner et stocker.
SQLite est un excellent premier choix pour ce « quelque chose » :
| Besoin | Comment SQLite y répond |
|---|---|
| Aucune installation | Intégré à la bibliothèque standard de Python (sqlite3) |
| Un seul fichier | Facile à sauvegarder, copier ou ouvrir dans un navigateur graphique |
| Requêtes SQL | Filtrage, regroupement, fonctions JSON, fonctions de date |
| Débits d'écriture modérés | Des milliers d'insertions par seconde lorsqu'elles sont groupées en transactions |
| Appareils edge | Tourne sans peine sur un Raspberry Pi ou un PC industriel |
Ce n'est pas le bon outil pour de nombreux écrivains concurrents sur différentes machines, ni pour des années de données à haute fréquence sur toute une flotte. Pour ces cas, consultez le guide MQTT vers TimescaleDB ou le stockage des données MQTT dans InfluxDB. Pour un site unique, une passerelle ou un prototype, SQLite est difficile à battre.
Architecture : garder le thread réseau libre
L'approche naïve exécute un INSERT et un commit() dans on_message. Cela fonctionne à quelques messages par minute, mais chaque commit force une synchronisation disque. À des débits plus élevés, le callback prend du retard, la boucle réseau de paho se bloque et les pings de keepalive peuvent être retardés au point que le broker coupe la connexion.
La conception robuste sépare les deux tâches :
- paho fait tourner sa boucle réseau dans un thread en arrière-plan (
loop_start()). on_messagese contente de placer le message dans unequeue.Queue.- Le thread principal possède la connexion SQLite, récupère les messages dans la file et les écrit par lots, avec un commit par lot ou une fois par seconde.
Cela contourne aussi la règle de SQLite selon laquelle une connexion doit être utilisée depuis le thread qui l'a créée.
Étape 1 : installer et préparer
Vous n'avez besoin que de paho-mqtt ; sqlite3 est fourni avec Python :
python -m venv .venv
source .venv/bin/activate # Windows : .venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Choisissez un préfixe de topic aléatoire pour que vos données ne se mélangent pas à celles des autres utilisateurs du serveur MQTT public gratuit, réservé aux tests et aux prototypes. Ce tutoriel utilise connectmq/demo/k7f3q2.
Étape 2 : le logger MQTT vers SQLite en Python
Enregistrez ceci sous mqtt_sqlite_logger.py :
import queue
import signal
import sqlite3
import time
import uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PORT = 1883
TOPIC = "connectmq/demo/k7f3q2/#"
DB_PATH = "mqtt_messages.db"
BATCH_SIZE = 200
FLUSH_INTERVAL = 1.0 # secondes
SCHEMA = """
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
received_at TEXT NOT NULL,
topic TEXT NOT NULL,
payload TEXT,
qos INTEGER NOT NULL,
retained INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_topic_time ON messages (topic, received_at);
CREATE INDEX IF NOT EXISTS idx_messages_time ON messages (received_at);
"""
inbox: "queue.Queue[tuple]" = queue.Queue(maxsize=50_000)
running = True
def utc_now() -> str:
# 2026-10-16T08:15:30.123Z, triable et compris par les fonctions de date de SQLite
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
def on_connect(client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
print(f"Connect failed: {reason_code}")
return
print(f"Connected, subscribing to {TOPIC}")
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
row = (
utc_now(),
msg.topic,
msg.payload.decode("utf-8", errors="replace"),
msg.qos,
int(msg.retain),
)
try:
inbox.put_nowait(row)
except queue.Full:
print("Queue full, dropping message") # la base de données ne suit pas
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
print(f"Disconnected: {reason_code}, paho will reconnect")
def open_db(path: str) -> sqlite3.Connection:
db = sqlite3.connect(path)
db.execute("PRAGMA journal_mode=WAL") # les lecteurs ne bloquent pas l'écrivain
db.execute("PRAGMA synchronous=NORMAL") # sûr avec WAL, beaucoup plus rapide
db.executescript(SCHEMA)
return db
def flush(db: sqlite3.Connection, batch: list) -> None:
if not batch:
return
with db: # une transaction par lot
db.executemany(
"INSERT INTO messages (received_at, topic, payload, qos, retained) VALUES (?, ?, ?, ?, ?)",
batch,
)
print(f"Stored {len(batch)} messages")
batch.clear()
def stop(*_):
global running
running = False
def main():
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)
db = open_db(DB_PATH)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sqlite-{uuid.uuid4().hex[:8]}")
# Broker ConnectMQ privé :
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set(); et utilisez le port 8883
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.connect_async(BROKER, PORT, keepalive=60)
client.loop_start()
batch: list = []
last_flush = time.monotonic()
try:
while running:
try:
batch.append(inbox.get(timeout=0.2))
except queue.Empty:
pass
if len(batch) >= BATCH_SIZE or time.monotonic() - last_flush >= FLUSH_INTERVAL:
flush(db, batch)
last_flush = time.monotonic()
finally:
client.disconnect()
client.loop_stop()
while not inbox.empty():
batch.append(inbox.get_nowait())
flush(db, batch)
db.close()
print("Logger stopped cleanly")
if __name__ == "__main__":
main()
Lancez-le avec python mqtt_sqlite_logger.py. Dans un autre terminal, publiez quelques données de test :
for i in $(seq 1 20); do
mosquitto_pub -h broker.connectmq.com -p 1883 -i sqlite-test-pub \
-t "connectmq/demo/k7f3q2/sensors/room$((i % 2 + 1))" \
-m "{\"temperature\": $((20 + RANDOM % 5)).$((RANDOM % 10)), \"humidity\": $((40 + RANDOM % 10))}"
sleep 1
done
Le logger affiche « Stored 1 messages » environ une fois par seconde. Arrêtez-le avec Ctrl+C : le reste de la file est vidé avant la sortie.
Comment le logger reste fiable
Horodatage. Le logger enregistre l'heure de réception du message, en UTC avec les millisecondes. Les horloges des appareils dérivent et certains appareils n'en ont pas : un horodatage de réception constitue donc une base fiable. Si votre payload porte son propre horodatage, stockez les deux. Le guide Python MQTT JSON montre comment en ajouter un côté publication.
Mode WAL. La journalisation write-ahead permet à un tableau de bord ou à la CLI sqlite3 de lire la base pendant que le logger écrit. Sans elle, de longues lectures peuvent bloquer les insertions.
File bornée. Si le disque cale, la file grossit jusqu'à 50 000 messages, puis rejette les nouveaux en affichant un message au lieu d'épuiser la mémoire. Ajustez la taille à votre trafic.
Reconnexions. connect_async() combiné à loop_start() réessaie en arrière-plan, et l'abonnement est rétabli dans on_connect. Les messages publiés pendant que le logger est hors ligne sont perdus, sauf si vous utilisez une session persistante. Avec MQTT 5, vous pouvez définir un intervalle d'expiration de session et clean_start=False pour que le broker mette les messages QoS 1 en file pour vous ; le serveur MQTT public gratuit conserve les sessions jusqu'à 30 minutes. Le guide de reconnexion paho-mqtt explique les détails.
Retained messages. À chaque reconnexion, le broker renvoie les retained messages : vous les verrez donc stockés plusieurs fois. La colonne retained vous permet de les exclure de l'analyse.
Étape 3 : interroger votre historique MQTT
Ouvrez la base avec la CLI SQLite (sqlite3 mqtt_messages.db) ou n'importe quel navigateur SQLite, et essayez ces requêtes.
Les 10 derniers messages d'un topic :
SELECT received_at, payload
FROM messages
WHERE topic = 'connectmq/demo/k7f3q2/sensors/room1'
ORDER BY received_at DESC
LIMIT 10;
La dernière valeur par topic, en s'appuyant sur le comportement documenté de SQLite pour les colonnes nues avec MAX() :
SELECT topic, payload, MAX(received_at) AS last_seen
FROM messages
WHERE retained = 0
GROUP BY topic;
Les moyennes horaires à partir de payloads JSON, avec les fonctions JSON intégrées de SQLite :
SELECT strftime('%Y-%m-%d %H:00', received_at) AS hour,
topic,
ROUND(AVG(json_extract(payload, '$.temperature')), 2) AS avg_temperature,
COUNT(*) AS samples
FROM messages
WHERE topic LIKE 'connectmq/demo/k7f3q2/sensors/%'
AND json_valid(payload)
GROUP BY hour, topic
ORDER BY hour;
Ne conserver que les sept derniers jours et récupérer l'espace :
DELETE FROM messages
WHERE received_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 days');
VACUUM;
Comme received_at est stocké dans un format ISO 8601 fixe, une simple comparaison de chaînes trie correctement et le filtre de plage temporelle peut utiliser l'index. Vous pouvez lancer la requête de rétention depuis cron ou depuis le logger lui-même une fois par jour.
Pour travailler sur les données en Python, pandas.read_sql_query("SELECT ...", sqlite3.connect("mqtt_messages.db")) renvoie un DataFrame. Le tutoriel MQTT et pandas couvre l'analyse et le rééchantillonnage.
Liste de vérification pour le dépannage
- « database is locked » : un autre processus détient une transaction d'écriture. Vérifiez que le mode WAL est activé et qu'un seul logger écrit dans le fichier.
- Rien n'est stocké : vérifiez que le filtre de topic correspond au topic du publisher ; testez avec
mosquitto_sub -h broker.connectmq.com -p 1883 -i check-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v. json_extractrenvoie NULL : le nom de la clé ou le chemin JSON diffère, ou le payload n'est pas du JSON. Inspectez d'abord quelques lignes brutes.- La base grossit vite : ajoutez la tâche de rétention et envisagez de ne publier que les valeurs modifiées côté publisher.
- Le logger perd des messages : augmentez
BATCH_SIZE, placez la base sur un stockage plus rapide ou limitez l'abonnement aux topics dont vous avez réellement besoin.
Du prototype à la production
Pour de vraies installations, faites tourner le logger comme un service (systemd sous Linux) afin qu'il redémarre après un reboot, et pointez-le vers un broker privé plutôt que vers le serveur MQTT public gratuit. Avec ConnectMQ, vous créez un broker dans le portail, copiez le nom d'utilisateur, le mot de passe et le client ID générés, et activez TLS sur le port 8883 ; le tutoriel Python MQTT TLS montre les réglages exacts. Le broker ConnectMQ ne stocke pas lui-même l'historique des messages : un subscriber externe comme celui-ci est donc le pattern honnête et recommandé. Si vous avez besoin d'un point de départ pour le code client, consultez le tutoriel Paho MQTT Python, et pour les autres langages la vue d'ensemble des bibliothèques clientes MQTT. La documentation Python de sqlite3 détaille les transactions et les options de connexion.
Démarrer avec ConnectMQ
ConnectMQ est un broker MQTT managé avec MQTT 3.1.1 et 5.0, TLS et WebSocket, ainsi qu'un explorateur de topics en direct pour vérifier ce qu'envoient vos appareils avant de le journaliser. Commencez à journaliser depuis broker.connectmq.com dès aujourd'hui, puis passez à votre propre broker MQTT dans le cloud. Créez un compte gratuit avec l'offre Serverless, sans carte bancaire.
Questions fréquentes
SQLite peut-il gérer un flux continu de données MQTT ?
Oui, pour un seul logger à des débits modérés. Grouper les insertions en transactions et activer le mode WAL permettent à SQLite d'absorber un flux régulier provenant de nombreux capteurs sur un matériel modeste.
Pourquoi ne pas insérer directement dans on_message ?
Effectuer un commit dans le callback bloque la boucle réseau de paho sur les entrées-sorties disque. Sous charge, cela retarde les pings de keepalive et les accusés de réception : une file d'attente et un écrivain séparé sont donc plus fiables.
Comment éviter de perdre des messages pendant le redémarrage du logger ?
Utilisez la QoS 1 avec une session MQTT 5 persistante (clean_start=False et un intervalle d'expiration de session) pour que le broker mette les messages en file pendant votre absence, dans la limite de sa durée d'expiration de session.
Peut-on stocker des payloads binaires dans SQLite ?
Oui. Changez la colonne payload en BLOB et insérez directement msg.payload au lieu de le décoder en texte.
