MQTT to SQLite with Python: Store and Query Sensor Messages

Sending data from MQTT to SQLite with Python is the simplest way to give your IoT project a history: one file, no database server, and full SQL for queries. In this tutorial you will build a paho-mqtt subscriber that stores every message from broker.connectmq.com with its timestamp, topic and flags, writes in efficient batches, and survives reconnects. Then you will query the data for recent values, hourly averages and retention cleanup.

Why SQLite for MQTT Data?

An MQTT broker delivers messages to whoever is subscribed right now. It is not a database, and apart from one retained message per topic it does not keep history. If you want to answer "what was the temperature at 3 a.m.?", something has to subscribe and store.

SQLite is a great first choice for that something:

It is not the right tool for many concurrent writers on different machines or for years of high-frequency data across a fleet. For those cases, look at the MQTT to TimescaleDB guide or storing MQTT data in InfluxDB. For a single site, a gateway or a prototype, SQLite is hard to beat.

Architecture: Keep the Network Thread Free

The naive approach executes an INSERT and a commit() inside on_message. That works at a few messages per minute, but each commit forces a disk sync. At higher rates the callback falls behind, paho's network loop stalls, and keepalive pings can be delayed long enough for the broker to drop the connection.

The robust design separates the two jobs:

1. paho runs its network loop in a background thread (loop_start()).
2. on_message does nothing but put the message on a queue.Queue.
3. The main thread owns the SQLite connection, pulls messages from the queue and writes them in batches, committing once per batch or once per second.

This also sidesteps SQLite's rule that a connection should be used from the thread that created it.

Step 1: Install and Prepare

You only need paho-mqtt; sqlite3 ships with Python:

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0"

Pick a random topic prefix so your data does not mix with other users of the public MQTT test broker. This tutorial uses connectmq/demo/k7f3q2.

Step 2: The MQTT to SQLite Python Logger

Save this as 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  # seconds

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, sortable and understood by SQLite date functions
    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")  # the database cannot keep up


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")    # readers do not block the writer
    db.execute("PRAGMA synchronous=NORMAL")  # safe with WAL, much faster
    db.executescript(SCHEMA)
    return db


def flush(db: sqlite3.Connection, batch: list) -> None:
    if not batch:
        return
    with db:  # one transaction per batch
        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]}")
    # Private ConnectMQ broker:
    # client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
    # client.tls_set(); and use 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()

Run it with python mqtt_sqlite_logger.py. In another terminal, publish some test data:

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

The logger prints "Stored 1 messages" about once a second. Stop it with Ctrl+C and the remaining queue is flushed before exit.

How the Logger Stays Reliable

Timestamps. The logger records when the message was received, in UTC with milliseconds. Device clocks drift and some devices have none, so a receive timestamp is a trustworthy baseline. If your payload carries its own timestamp, store both. The Python MQTT JSON guide shows how to add one on the publishing side.

WAL mode. Write-ahead logging lets a dashboard or the sqlite3 CLI read the database while the logger writes. Without it, long reads can block inserts.

Bounded queue. If the disk stalls, the queue grows up to 50,000 messages and then drops new ones with a message instead of exhausting memory. Tune the size to your traffic.

Reconnects. connect_async() plus loop_start() keeps retrying in the background, and the subscription is restored in on_connect. Messages published while the logger is offline are lost unless you use a persistent session. With MQTT 5 you can set a session expiry interval and clean_start=False so the broker queues QoS 1 messages for you; the public broker keeps sessions for up to 30 minutes. The paho-mqtt reconnect guide explains the details.

Retained messages. On every reconnect the broker re-sends retained messages, so you will see them stored more than once. The retained column lets you filter them out of analysis.

Step 3: Query Your MQTT History

Open the database with the SQLite CLI (sqlite3 mqtt_messages.db) or any SQLite browser, and try these queries.

Latest 10 messages for a topic:

SELECT received_at, payload
FROM messages
WHERE topic = 'connectmq/demo/k7f3q2/sensors/room1'
ORDER BY received_at DESC
LIMIT 10;

Latest value per topic, using SQLite's documented behaviour for bare columns with MAX():

SELECT topic, payload, MAX(received_at) AS last_seen
FROM messages
WHERE retained = 0
GROUP BY topic;

Hourly averages from JSON payloads, using SQLite's built-in JSON functions:

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;

Keep only the last seven days and reclaim space:

DELETE FROM messages
WHERE received_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 days');
VACUUM;

Because received_at is stored in a fixed ISO 8601 format, plain string comparison sorts correctly and the time-range filter can use the index. You can run the retention query from cron or from the logger itself once a day.

To work with the data in Python, pandas.read_sql_query("SELECT ...", sqlite3.connect("mqtt_messages.db")) returns a DataFrame. The MQTT and pandas tutorial covers analysis and resampling.

Troubleshooting Checklist

1. "database is locked": another process holds a write transaction. Make sure WAL mode is enabled and only one logger writes to the file.
2. Nothing stored: confirm the topic filter matches the publisher's topic; test with mosquitto_sub -h broker.connectmq.com -p 1883 -i check-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v.
3. json_extract returns NULL: the key name or JSON path differs, or the payload is not JSON. Inspect a few raw rows first.
4. Database grows quickly: add the retention job and consider storing only changed values on the publisher side.
5. Logger drops messages: increase BATCH_SIZE, move the database to faster storage, or reduce the subscription to the topics you actually need.

From Prototype to Production

For real installations, run the logger as a service (systemd on Linux) so it restarts after reboots, and point it at a private broker instead of the public one. With ConnectMQ, you create a broker in the portal, copy the generated username, password and client ID, and enable TLS on port 8883; the Python MQTT TLS tutorial shows the exact settings. The ConnectMQ broker does not store message history itself, so an external subscriber like this one is the honest and recommended pattern. If you need a starting point for the client code, see the paho mqtt python tutorial, and for other languages the MQTT client libraries overview. The Python sqlite3 documentation covers transactions and connection options in depth.

Get Started with ConnectMQ

ConnectMQ is a hosted MQTT broker with MQTT 3.1.1 and 5.0, TLS and WebSocket, plus a live topic explorer for checking what your devices send before you log it. Start logging from broker.connectmq.com today, then move to your own MQTT broker in the cloud. Create a free account on the Serverless plan, no credit card required.

Frequently Asked Questions

Can SQLite handle continuous MQTT data?

Yes, for a single logger at moderate rates. Batching inserts into transactions and enabling WAL mode lets SQLite absorb a steady stream from many sensors on modest hardware.

Why not insert directly inside on_message?

Committing in the callback blocks paho's network loop on disk I/O. Under load that delays keepalive pings and acknowledgments, so a queue and a separate writer is more reliable.

How do I avoid losing messages while the logger restarts?

Use QoS 1 with a persistent MQTT 5 session (clean_start=False and a session expiry interval) so the broker queues messages while you are offline, within its session expiry limit.

Can I store binary payloads in SQLite?

Yes. Change the payload column to BLOB and insert msg.payload directly instead of decoding it to text.