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.

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.