EdgeSync

A lightweight Python library for reliable data delivery from edge applications to cloud services. EdgeSync provides persistent local queuing, store-and-forward synchronization, automatic retries, and recovery from network or application failures. Built for IoT devices, edge gateways, industrial systems, and any application operating with intermittent connectivity.

Why EdgeSync

Edge applications lose data for the same handful of reasons every time: the network drops, the cloud endpoint is temporarily unavailable, or the process crashes mid-send. EdgeSync solves this with a durable store-and-forward architecture:

1. Data is persisted locally (SQLite) before anything is sent.
2. A background worker attempts delivery to the configured destination.
3. Acknowledged messages are removed from the local queue.
4. Failed messages stay queued and are retried with backoff.
5. Pending data survives process crashes and device restarts.

EdgeSync provides at-least-once delivery, not exactly-once — see docs/reliability.md for the exact guarantee and how to build idempotent consumers on top of it.

Install

pip install edgesync

Quickstart

import asyncio
from edgesync import EdgeSync


async def main():
    async with EdgeSync(
        database="edgesync.db",
        endpoint="https://api.example.com/telemetry",
    ) as sync:
        receipt = await sync.publish(
            {
                "device_id": "device-001",
                "temperature": 28.5,
            }
        )
        print(f"queued as {receipt.message_id}")


asyncio.run(main())

That's it: publish() durably persists the message and returns as soon as it's on disk. A background worker delivers it, retrying with exponential backoff on failure, without the caller needing to stay connected or wait for the network.

MQTT

Install the optional extra first:

pip install edgesync[mqtt]

Publishing to MQTT instead of HTTP is the same pattern with a different transport:

from edgesync import EdgeSync
from edgesync.transports.mqtt import MQTTTransport

sync = EdgeSync(
    database="edgesync.db",
    destinations={
        "telemetry": MQTTTransport(
            "broker.example.com",
            "devices/device-001/telemetry",
            qos=1,
            username="device-001",
            password="...",
        ),
    },
    default_destination="telemetry",
)

await sync.publish({"temperature": 28.5})

MQTTTransport only accepts QoS 1 or 2 — QoS 0 publishes are never acknowledged by the broker, so there'd be no way to tell a successful delivery from a lost one, which would break EdgeSync's at-least-once guarantee. Unlike HTTPTransport, MQTT holds a persistent broker connection: EdgeSync.start() still won't raise if the broker is unreachable (data keeps queuing locally either way), and the transport automatically reconnects on the next delivery attempt after a dropped connection or failed publish.

TLS: by default MQTTTransport connects in plaintext on port 1883. To connect over TLS, pass tls_params (an aiomqtt.TLSParameters) and switch to the broker's TLS port, conventionally 8883:

import aiomqtt
from edgesync.transports.mqtt import MQTTTransport

MQTTTransport(
    "broker.example.com",
    "devices/device-001/telemetry",
    port=8883,
    tls_params=aiomqtt.TLSParameters(
        ca_certs="/etc/ssl/certs/ca-certificates.crt",
    ),
)

TLSParameters also accepts certfile / keyfile for client-certificate (mutual TLS) authentication. Omit tls_params entirely for an unencrypted connection, e.g. a broker on a trusted local network or a local development/test setup. See the MQTT section of the getting-started guide for further detail.

Usage with FastAPI, Flask, and plain asyncio

EdgeSync is async-native, so it's most at home in an asyncio app or an async framework like FastAPI — start it once, publish from your handlers, close it on shutdown:

# FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
    await sync.start()
    yield
    await sync.close()


app = FastAPI(lifespan=lifespan)


@app.post("/telemetry")
async def telemetry(payload: dict):
    receipt = await sync.publish(payload)
    return {"message_id": receipt.message_id}

Flask is synchronous, so it needs a small bridge to run EdgeSync's event loop on a background thread rather than spinning up a new loop per request. See docs/integrations.md for the full FastAPI, Flask, and plain-asyncio guide, including the ready-to-use Flask bridge.

Features

• Durable local queue — SQLite-backed, survives crashes and restarts.
• At-least-once delivery — messages are only removed once the destination acknowledges them.
• Automatic retries — configurable exponential backoff with jitter.
• Dead-letter queue — messages that exhaust retries or fail permanently are retained for inspection instead of being silently dropped.
• Multiple destinations — route different message types to different endpoints or transports.
• Pluggable transports — ships with HTTP and MQTT transports; implement Transport for anything else (gRPC, a message broker, ...).
• Bounded storage — configurable queue capacity with a choice of overflow policies.
• Async-native — built on asyncio and httpx.

Documentation

• Getting started
• Using EdgeSync with FastAPI, Flask, and plain asyncio
• Reliability & delivery guarantees
• Storage design

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy

See CONTRIBUTING.md for the full workflow.

License

MIT — see LICENSE.