Building a Resilient Offline-First Telemetry Collector in Python for Edge IoT Devices

#python #iot #architecture #backend

Why Edge Devices Need Offline Resilience

When deploying IoT sensors in remote environments such as greenhouses or agricultural fields, internet connectivity is rarely guaranteed. If a sensor node fails to reach the cloud database, critical telemetry data like temperature, pH, or electrical conductivity can be permanently lost.

In this tutorial, we will build a lightweight Python telemetry collector that buffers sensor readings locally in SQLite during network outages and automatically flushes them to a remote API once connectivity is restored.

Architecture Overview

Ingestion Layer: Reads incoming JSON payloads from local microcontrollers (such as an ESP32 over HTTP or MQTT).
Local Buffer (SQLite): Acts as a persistent write-ahead queue when the upstream cloud endpoint is unreachable.
Sync Worker: A background loop that checks network health and batches pending records to the cloud.

Setting Up the Local Buffer

First, let's create a simple SQLite schema to store unsent telemetry payloads safely on disk:

import sqlite3
import json
import time

DB_PATH = "telemetry_buffer.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS pending_metrics (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT NOT NULL,
            payload TEXT NOT NULL,
            created_at REAL NOT NULL
        )
    """)
    conn.commit()
    conn.close()

def save_locally(device_id: str, metrics: dict):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO pending_metrics (device_id, payload, created_at) VALUES (?, ?, ?)",
        (device_id, json.dumps(metrics), time.time())
    )
    conn.commit()
    conn.close()

Implementing the Batch Sync Worker

Next, we create a worker function that attempts to push buffered readings to our cloud ingestion endpoint. If the request fails due to a timeout or connection error, the records remain safely stored for the next retry cycle.

import urllib.request
import urllib.error

CLOUD_ENDPOINT = "https://example.com"

def flush_buffer(batch_size: int = 50):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT id, device_id, payload, created_at FROM pending_metrics ORDER BY id ASC LIMIT ?",
        (batch_size,)
    )
    rows = cursor.fetchall()

    if not rows:
        conn.close()
        return

    batch = [
        {
            "id": row[0],
            "device_id": row[1],
            "metrics": json.loads(row[2]),
            "timestamp": row[3]
        }
        for row in rows
    ]

    try:
        req = urllib.request.Request(
            CLOUD_ENDPOINT,
            data=json.dumps({"records": batch}).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST"
        )
        with urllib.request.urlopen(req, timeout=5) as response:
            if response.status in (200, 201):
                ids_to_delete = [(row[0],) for row in rows]
                cursor.executemany("DELETE FROM pending_metrics WHERE id = ?", ids_to_delete)
                conn.commit()
                print(f"Successfully synced {len(rows)} records.")
    except urllib.error.URLError as e:
        print(f"Network unavailable, keeping records in buffer: {e}")
    finally:
        conn.close()

Key Takeaways

Zero Data Loss: Using SQLite as a local queue guarantees that transient Wi-Fi drops never erase historical sensor trends.
Batch Efficiency: Sending 50 records in a single HTTP POST reduces TLS handshake overhead and lowers cloud API costs.
Idempotency: Always include the original created_at edge timestamp so backend dashboards plot metrics when they actually occurred, not when they arrived.

DEV Community