Buffer MQTT Messages Offline in Python with SQLite (Store and Forward)

To buffer MQTT messages offline in Python, write every reading to a local SQLite queue first and let a separate sender drain that queue to the broker only when the connection is up. In this tutorial you will build that store-and-forward publisher with paho-mqtt 2.x: it survives network outages and reboots, replays messages in the original order, deletes them only after the broker acknowledges them and keeps disk usage bounded. It runs on a Raspberry Pi or any Linux machine.

Why paho-mqtt's Built-in Queue Is Not Enough

paho-mqtt already queues outgoing messages while it reconnects, so why build anything? Because that queue lives in memory. If the process restarts, the Pi reboots or power is cut during an outage, everything in it is gone. For a gateway on cellular, a site with unreliable Wi-Fi or a vehicle that drives through tunnels, those are exactly the moments that matter.

There is also a subtle problem with how the broker treats publishers. Persistent sessions and QoS 1 or 2 protect messages to subscribers that are temporarily offline. They do nothing for a publisher that cannot reach the broker at all. Durability on the publishing side is the client's job.

The design that solves both problems is a classic outbox:

1. The application never publishes directly. It inserts a row into an SQLite table: topic, payload, QoS, retain flag and timestamp.
2. A sender thread reads the oldest rows, publishes them at QoS 1 and waits for the broker's PUBACK.
3. Only acknowledged rows are deleted. Anything else stays for the next attempt.

Because SQLite is a file, the queue survives restarts. Because the sender processes rows in insertion order over a single connection, order is preserved.

Designing the Outbox Table

outbox
  id        INTEGER PRIMARY KEY AUTOINCREMENT   -- defines replay order
  topic     TEXT NOT NULL
  payload   BLOB NOT NULL
  qos       INTEGER NOT NULL DEFAULT 1
  retain    INTEGER NOT NULL DEFAULT 0
  created   REAL NOT NULL                      -- unix time of the reading

Two SQLite settings make a big difference on a Pi. journal_mode=WAL lets the producer insert while the sender reads, and reduces the number of writes per transaction. synchronous=NORMAL is a reasonable balance between safety and SD card wear in WAL mode. Put the database on an SSD or a writable partition if you run a read-only root filesystem.

The Store-and-Forward Publisher

The module below is self-contained. Import OfflineBufferedPublisher, call publish() from anywhere in your application and it takes care of the rest.

import json
import os
import sqlite3
import threading
import time
import uuid

import paho.mqtt.client as mqtt


class OfflineBufferedPublisher:
    def __init__(self, host, port=1883, client_id=None, username=None, password=None,
                 db_path="outbox.db", max_rows=200_000, batch=20, max_rate=8.0):
        self.db_path = db_path
        self.max_rows = max_rows
        self.batch = batch
        self.min_interval = 1.0 / max_rate if max_rate else 0
        self._lock = threading.Lock()
        self._wake = threading.Event()
        self._stop = threading.Event()

        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.execute("PRAGMA synchronous=NORMAL")
        self.db.execute("""CREATE TABLE IF NOT EXISTS outbox (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT NOT NULL, payload BLOB NOT NULL,
            qos INTEGER NOT NULL DEFAULT 1, retain INTEGER NOT NULL DEFAULT 0,
            created REAL NOT NULL)""")
        self.db.commit()

        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=client_id or f"buffered-{uuid.uuid4().hex[:8]}")
        if username:
            self.client.username_pw_set(username, password)
        if port == 8883:
            self.client.tls_set()
        self.client.reconnect_delay_set(1, 60)
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.connect_async(host, port, keepalive=30)
        self.client.loop_start()

        self.sender = threading.Thread(target=self._drain, daemon=True)
        self.sender.start()

    # ---- producer side -------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if not isinstance(payload, (bytes, bytearray)):
            payload = json.dumps(payload).encode() if not isinstance(payload, str) else payload.encode()
        with self._lock:
            self.db.execute("INSERT INTO outbox (topic, payload, qos, retain, created) VALUES (?,?,?,?,?)",
                            (topic, payload, qos, int(retain), time.time()))
            self._enforce_limit()
            self.db.commit()
        self._wake.set()

    def pending(self):
        with self._lock:
            return self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]

    def _enforce_limit(self):
        count = self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
        if count > self.max_rows:
            self.db.execute("DELETE FROM outbox WHERE id IN (SELECT id FROM outbox ORDER BY id LIMIT ?)",
                            (count - self.max_rows,))

    # ---- sender side ---------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        print("connected:", reason_code)
        self._wake.set()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        print("disconnected:", reason_code)

    def _drain(self):
        while not self._stop.is_set():
            if not self.client.is_connected():
                self._wake.wait(2)
                self._wake.clear()
                continue
            with self._lock:
                rows = self.db.execute(
                    "SELECT id, topic, payload, qos, retain FROM outbox ORDER BY id LIMIT ?",
                    (self.batch,)).fetchall()
            if not rows:
                self._wake.wait(5)
                self._wake.clear()
                continue
            for row_id, topic, payload, qos, retain in rows:
                info = self.client.publish(topic, payload, qos=qos, retain=bool(retain))
                try:
                    info.wait_for_publish(timeout=10)
                except (ValueError, RuntimeError):
                    break
                if not info.is_published():
                    break  # connection dropped; keep this row and everything after it
                with self._lock:
                    self.db.execute("DELETE FROM outbox WHERE id = ?", (row_id,))
                    self.db.commit()
                if self.min_interval:
                    time.sleep(self.min_interval)

    def close(self):
        self._stop.set()
        self._wake.set()
        self.sender.join(timeout=5)
        self.client.loop_stop()
        self.client.disconnect()
        self.db.close()


if __name__ == "__main__":
    pub = OfflineBufferedPublisher(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        client_id=os.getenv("MQTT_CLIENT_ID"),
        username=os.getenv("MQTT_USERNAME"),
        password=os.getenv("MQTT_PASSWORD"),
    )
    prefix = os.getenv("TOPIC_PREFIX", "connectmq/demo/q5w8e1/sensor1")
    n = 0
    try:
        while True:
            n += 1
            pub.publish(f"{prefix}/reading", {"seq": n, "ts": round(time.time(), 3), "value": 20 + n % 5})
            print("queued", n, "pending", pub.pending())
            time.sleep(2)
    except KeyboardInterrupt:
        pub.close()

Testing an Outage

Start a subscriber that prints sequence numbers, then run the publisher:

mosquitto_sub -h broker.connectmq.com -p 1883 -i buffer-viewer-q5w8e1 -t "connectmq/demo/q5w8e1/#" -v
python buffered_publisher.py

Now simulate an outage. On a Pi you can drop the network with sudo ip link set wlan0 down (do this from a local console, not over SSH on the same interface), wait a minute, then bring it back with sudo ip link set wlan0 up. The "pending" counter grows during the outage. When paho reconnects, the sender drains the backlog and the subscriber receives every missing sequence number in order, followed by live readings. Try stopping the script during the outage and starting it again: the backlog is still there.

Ordering, Duplicates and Timestamps

Order. The sender publishes one message at a time and waits for its acknowledgement before deleting it and moving on. Over a single MQTT connection, that preserves the original order. You can raise throughput by allowing several in-flight messages, but keep a single sender thread and a single connection if order matters.

Duplicates. QoS 1 is at-least-once. If the connection drops after the broker received a message but before the PUBACK arrived, the row is sent again after reconnecting. Include a sequence number or a unique ID in the payload so consumers can discard duplicates. MQTT QoS levels explains the trade-offs, and why QoS 2 still does not protect you from a client restart.

Timestamps. A replayed message arrives minutes or hours late. Always put the measurement time in the payload, and have consumers use it rather than the arrival time. With MQTT 5 you can also set a message expiry interval on readings that become meaningless after a while, such as a live position.

Tuning for Real Deployments

When the queue is full, this implementation drops the oldest rows, which suits telemetry where recent data matters most. For event logs you might prefer to stop accepting new rows and raise an alarm instead. Either way, publish the pending() count as a health metric, so a growing backlog is visible on your dashboard.

Run the publisher as a systemd service so it restarts with the device, as shown in running a Python MQTT script on boot. For other ways of storing MQTT data locally, see MQTT to SQLite in Python, which covers the subscriber side.

Alternative: A Local Broker with a Bridge

Another common store-and-forward design puts a Mosquitto broker on the Pi itself. Local applications publish to localhost, and Mosquitto bridges the topics to the cloud broker, queuing messages while the uplink is down. It is a good fit when several processes on the same device publish data, or when you cannot change application code. The Mosquitto bridge configuration guide covers the settings.

The trade-offs are worth knowing. You now operate a broker on every gateway, with its own configuration, updates and credentials. Mosquitto's queue limits and persistence behaviour need careful tuning to survive reboots, and debugging data gaps means looking at two brokers instead of one. The Python outbox keeps everything inside one process that you fully control and can test with a unit test, which many teams find simpler for a single-purpose gateway. Choose the bridge when many local producers share one uplink, and the outbox when one application owns the data.

Production Broker Settings

The free public MQTT broker is fine for this test, but everything published there is public. For real data, create a private broker on ConnectMQ, then start the publisher with MQTT_PORT=8883, MQTT_USERNAME=YOUR_USERNAME, MQTT_PASSWORD=YOUR_PASSWORD and MQTT_CLIENT_ID=YOUR_CLIENT_ID from the portal. A managed MQTT broker takes care of uptime on the server side, while this outbox covers the gaps on the client side. The paho-mqtt documentation describes MQTTMessageInfo and the reconnect options in detail.

This pattern is a core building block for every gateway in the Raspberry Pi MQTT tutorial series, from CAN bus readers to LoRa receivers.

Never Lose a Reading with ConnectMQ

Combine a durable local outbox with a reliable broker and outages become invisible to your dashboards. Create a free Serverless broker at app.connectmq.com, no credit card required, point this publisher at it over TLS and watch the backlog drain into the ConnectMQ live topic explorer after your next network drop.

Frequently Asked Questions

Does MQTT persistent session buffer messages for a publisher?

No. Persistent sessions keep messages for subscribers that are offline. A publisher that cannot reach the broker must buffer locally, which is what this outbox does.

Will replayed messages arrive in the right order?

Yes, as long as one sender publishes rows in ID order over one connection and waits for each acknowledgement, as in this code.

Can I use this with asyncio?

Yes. The same outbox design works with aiomqtt: run the drain loop as an async task and use aiosqlite or run SQLite calls in a thread executor.

How do I avoid wearing out the SD card?

Use WAL mode, keep payloads compact and place the database on an SSD where possible. Writing a reading every few seconds is a modest load for good storage.