MQTT para SQLite com Python: Armazene e Consulte Mensagens de Sensores

Enviar dados de MQTT para SQLite com Python é a forma mais simples de dar histórico ao seu projeto de IoT: um único arquivo, nenhum servidor de banco de dados e SQL completo para as consultas. Neste tutorial você vai construir um subscriber paho-mqtt que armazena cada mensagem do servidor MQTT público (gratuito) broker.connectmq.com com timestamp, topic e flags, grava em lotes eficientes e sobrevive a reconexões. Depois, vai consultar os dados para obter valores recentes, médias por hora e fazer a limpeza por retenção.

Por que SQLite para dados MQTT?

Um broker MQTT entrega mensagens a quem está assinado naquele momento. Ele não é um banco de dados e, tirando uma retained message por topic, não guarda histórico. Se você quer responder "qual era a temperatura às 3 da manhã?", algo precisa assinar e armazenar.

O SQLite é uma ótima primeira escolha para esse "algo":

Ele não é a ferramenta certa para muitos escritores simultâneos em máquinas diferentes, nem para anos de dados de alta frequência de uma frota inteira. Para esses casos, veja o guia de MQTT para TimescaleDB ou como armazenar dados MQTT no InfluxDB. Para um único local, um gateway ou um protótipo, o SQLite é difícil de superar.

Arquitetura: mantenha a thread de rede livre

A abordagem ingênua executa um INSERT e um commit() dentro do on_message. Isso funciona com poucas mensagens por minuto, mas cada commit força uma sincronização em disco. Em taxas maiores, o callback fica para trás, o network loop do paho trava e os pings de keepalive podem atrasar o suficiente para o broker derrubar a conexão.

O design robusto separa os dois trabalhos:

1. O paho executa o network loop em uma thread em segundo plano (loop_start()).
2. O on_message apenas coloca a mensagem em uma queue.Queue.
3. A thread principal é dona da conexão SQLite, retira as mensagens da fila e as grava em lotes, fazendo commit uma vez por lote ou uma vez por segundo.

Isso também contorna a regra do SQLite de que uma conexão deve ser usada pela thread que a criou.

Passo 1: instalar e preparar

Você só precisa do paho-mqtt; o sqlite3 já vem com o Python:

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

Escolha um prefixo de topic aleatório para que os seus dados não se misturem com os de outros usuários do servidor MQTT público, que serve apenas para testes. Este tutorial usa connectmq/demo/k7f3q2.

Passo 2: o logger de MQTT para SQLite em Python

Salve isto como 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  # segundos

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, ordenável e compreendido pelas funções de data do 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")  # o banco de dados não está dando conta


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")    # leitores não bloqueiam o escritor
    db.execute("PRAGMA synchronous=NORMAL")  # seguro com WAL, muito mais rápido
    db.executescript(SCHEMA)
    return db


def flush(db: sqlite3.Connection, batch: list) -> None:
    if not batch:
        return
    with db:  # uma transação por lote
        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 privado:
    # client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
    # client.tls_set(); e use a porta 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()

Execute com python mqtt_sqlite_logger.py. Em outro terminal, publique alguns dados de teste:

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

O logger imprime "Stored 1 messages" cerca de uma vez por segundo. Pare-o com Ctrl+C e o que restar na fila é gravado antes de sair.

Como o logger se mantém confiável

Timestamps. O logger registra o momento em que a mensagem foi recebida, em UTC e com milissegundos. Relógios de dispositivos sofrem desvios e alguns dispositivos nem têm relógio, então um timestamp de recebimento é uma referência confiável. Se o seu payload traz o próprio timestamp, armazene os dois. O guia de JSON com MQTT em Python mostra como adicionar um no lado do publisher.

Modo WAL. O write-ahead logging permite que um dashboard ou a CLI sqlite3 leiam o banco enquanto o logger escreve. Sem ele, leituras longas podem bloquear os inserts.

Fila limitada. Se o disco travar, a fila cresce até 50.000 mensagens e então descarta as novas com um aviso, em vez de esgotar a memória. Ajuste o tamanho ao seu tráfego.

Reconexões. connect_async() com loop_start() continua tentando em segundo plano, e a assinatura é restaurada no on_connect. Mensagens publicadas enquanto o logger está offline se perdem, a menos que você use uma sessão persistente. Com MQTT 5, você pode definir um session expiry interval e clean_start=False para que o broker enfileire as mensagens QoS 1 para você; o servidor MQTT público mantém sessões por até 30 minutos. O guia de reconexão do paho-mqtt explica os detalhes.

Retained messages. A cada reconexão o broker reenvia as retained messages, então você as verá armazenadas mais de uma vez. A coluna retained permite filtrá-las da análise.

Passo 3: consulte o seu histórico MQTT

Abra o banco de dados com a CLI do SQLite (sqlite3 mqtt_messages.db) ou qualquer navegador de SQLite e experimente estas consultas.

As 10 mensagens mais recentes de um topic:

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

Último valor por topic, usando o comportamento documentado do SQLite para colunas simples com MAX():

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

Médias por hora a partir de payloads JSON, usando as funções JSON nativas do 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;

Mantenha apenas os últimos sete dias e recupere espaço:

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

Como received_at é armazenado em um formato ISO 8601 fixo, a simples comparação de strings ordena corretamente e o filtro por intervalo de tempo pode usar o índice. Você pode executar a consulta de retenção via cron ou pelo próprio logger uma vez por dia.

Para trabalhar com os dados em Python, pandas.read_sql_query("SELECT ...", sqlite3.connect("mqtt_messages.db")) retorna um DataFrame. O tutorial de MQTT com pandas aborda análise e reamostragem.

Checklist de solução de problemas

1. "database is locked": outro processo mantém uma transação de escrita. Garanta que o modo WAL esteja ativado e que apenas um logger escreva no arquivo.
2. Nada é armazenado: confirme que o filtro de topic corresponde ao topic do publisher; teste com mosquitto_sub -h broker.connectmq.com -p 1883 -i check-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v.
3. json_extract retorna NULL: o nome da chave ou o caminho JSON é diferente, ou o payload não é JSON. Inspecione algumas linhas brutas primeiro.
4. O banco cresce rápido: adicione a rotina de retenção e considere armazenar apenas valores alterados no lado do publisher.
5. O logger descarta mensagens: aumente BATCH_SIZE, mova o banco para um armazenamento mais rápido ou reduza a assinatura aos topics de que você realmente precisa.

Do protótipo à produção

Em instalações reais, rode o logger como serviço (systemd no Linux) para que ele reinicie após reboots, e aponte-o para um broker privado em vez do servidor MQTT público. Com o ConnectMQ, você cria um broker no portal, copia o usuário, a senha e o client ID gerados e ativa TLS na porta 8883; o tutorial de TLS com MQTT em Python mostra as configurações exatas. O broker do ConnectMQ não armazena histórico de mensagens por conta própria, então um subscriber externo como este é o padrão honesto e recomendado. Se você precisa de um ponto de partida para o código do cliente, veja o tutorial de Paho MQTT em Python e, para outras linguagens, a visão geral de bibliotecas de cliente MQTT. A documentação do sqlite3 do Python cobre transações e opções de conexão em profundidade.

Comece a usar o ConnectMQ

O ConnectMQ é um broker MQTT gerenciado com MQTT 3.1.1 e 5.0, TLS e WebSocket, além de um explorador de topics ao vivo para conferir o que os seus dispositivos enviam antes de registrar os dados. Comece a registrar a partir do broker.connectmq.com hoje e depois migre para o seu próprio broker MQTT na nuvem. Crie uma conta gratuita no plano Serverless, sem cartão de crédito.

Perguntas frequentes

O SQLite aguenta dados MQTT contínuos?

Sim, para um único logger em taxas moderadas. Agrupar os inserts em transações e ativar o modo WAL permite que o SQLite absorva um fluxo constante de muitos sensores em um hardware modesto.

Por que não inserir direto dentro do on_message?

Fazer commit no callback bloqueia o network loop do paho em I/O de disco. Sob carga, isso atrasa os pings de keepalive e as confirmações, então uma fila com um escritor separado é mais confiável.

Como evitar perder mensagens enquanto o logger reinicia?

Use QoS 1 com uma sessão persistente do MQTT 5 (clean_start=False e um session expiry interval), para que o broker enfileire as mensagens enquanto você está offline, dentro do limite de expiração de sessão.

Posso armazenar payloads binários no SQLite?

Sim. Mude a coluna payload para BLOB e insira msg.payload diretamente, em vez de decodificá-lo para texto.