SpatiaLite & GeoPackage Automation with Python

How to Build an Offline Sync Push Script for GeoPackage

Read the change-log for rows newer than the device's watermark, package them as a delta with a stable id, then apply the whole delta to the central…

Read the change-log for rows newer than the device’s watermark, package them as a delta with a stable id, then apply the whole delta to the central GeoPackage inside one BEGIN IMMEDIATE transaction — recording the delta id in an _applied_deltas ledger so a retry skips work already done, and wrapping the write in bounded lock-retry. This page, part of the Offline-First Sync Strategies guide, gives you that push script end to end using only the standard-library sqlite3 module.

Why This Matters

A push script runs unattended, over a flaky link, on a device that can be killed mid-write at any moment. If it is not idempotent, a retried push duplicates features; if it is not transactional, a crash leaves the central store half-updated; if it does not handle database is locked, it fails the instant another writer touches the file. The script below is built around exactly those three failure modes so an interrupted run is always safe to repeat.

This page focuses on the apply and idempotency logic. The connection-level retry mechanics — backoff strategy, jitter, and when to give up — are covered in depth in implementing connection retries for offline apps; here we use a compact retry wrapper and concentrate on the delta and the transaction.

Prerequisites

  • Python 3.9+ with the standard-library sqlite3 module (no third-party driver needed)
  • A field GeoPackage carrying the change-log and version counter described in the tracking guide
  • A central GeoPackage with the same parcels schema and a stable fid primary key
  • SQLite 3.35+ for reliable ON CONFLICT upsert semantics
  • Both files opened in WAL mode so a background push does not block collection

Primary Method

The complete push script reads a delta from the field file and applies it to the central file. It is idempotent, transactional, and retries on lock.

python
# Python 3.9+ — offline sync push script for GeoPackage (sqlite3 only)
import sqlite3
import time


def _connect(path: str) -> sqlite3.Connection:
    """Open a GeoPackage with WAL and a busy timeout, autocommit off via manual txn."""
    conn = sqlite3.connect(path, isolation_level=None, timeout=30.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA foreign_keys=ON;")
    return conn


def build_delta(field_path: str, table: str = "parcels") -> dict:
    """Collect changes above the device watermark into a self-contained delta."""
    conn = _connect(field_path)
    try:
        watermark = conn.execute(
            "SELECT last_pushed_version FROM _sync_state WHERE id = 1"
        ).fetchone()[0]
        rows = conn.execute(
            "SELECT feature_id, op, MAX(version) AS version "
            "FROM _sync_log WHERE table_name = ? AND version > ? "
            "GROUP BY feature_id",
            (table, watermark),
        ).fetchall()

        max_version = watermark
        upserts, tombstones = [], []
        for r in rows:
            max_version = max(max_version, r["version"])
            if r["op"] == "D":
                tombstones.append(r["feature_id"])
                continue
            feat = conn.execute(
                f"SELECT fid, name, status, AsGeoJSON(geom) AS geom "
                f"FROM {table} WHERE fid = ?",
                (r["feature_id"],),
            ).fetchone()
            if feat is not None:
                upserts.append(dict(feat))
    finally:
        conn.close()

    return {
        "delta_id": f"{table}:{watermark + 1}-{max_version}",
        "table": table,
        "base_version": watermark,
        "max_version": max_version,
        "upserts": upserts,
        "tombstones": tombstones,
    }


def _with_lock_retry(fn, attempts: int = 5, base_delay: float = 0.5):
    """Retry fn() on 'database is locked' with exponential backoff."""
    for attempt in range(attempts):
        try:
            return fn()
        except sqlite3.OperationalError as exc:
            if "locked" not in str(exc).lower() or attempt == attempts - 1:
                raise
            time.sleep(base_delay * (2 ** attempt))
    raise RuntimeError("unreachable")


def apply_delta(central_path: str, delta: dict) -> str:
    """Apply a delta once, inside one transaction; return 'applied' or 'skipped'."""
    conn = _connect(central_path)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS _applied_deltas ("
        "  delta_id TEXT PRIMARY KEY,"
        "  applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))"
        ")"
    )

    def _txn() -> str:
        conn.execute("BEGIN IMMEDIATE")
        try:
            seen = conn.execute(
                "SELECT 1 FROM _applied_deltas WHERE delta_id = ?",
                (delta["delta_id"],),
            ).fetchone()
            if seen:
                conn.execute("COMMIT")
                return "skipped"

            table = delta["table"]
            for f in delta["upserts"]:
                conn.execute(
                    f"INSERT INTO {table} (fid, name, status, geom) "
                    f"VALUES (?, ?, ?, GeomFromGeoJSON(?)) "
                    f"ON CONFLICT(fid) DO UPDATE SET "
                    f"  name = excluded.name, status = excluded.status, "
                    f"  geom = excluded.geom",
                    (f["fid"], f["name"], f["status"], f["geom"]),
                )
            for fid in delta["tombstones"]:
                conn.execute(f"DELETE FROM {table} WHERE fid = ?", (fid,))
            conn.execute(
                "INSERT INTO _applied_deltas (delta_id) VALUES (?)",
                (delta["delta_id"],),
            )
            conn.execute("COMMIT")
            return "applied"
        except Exception:
            conn.execute("ROLLBACK")
            raise

    try:
        return _with_lock_retry(_txn)
    finally:
        conn.close()


def advance_watermark(field_path: str, new_version: int) -> None:
    """Move the device watermark forward only after a confirmed apply."""
    conn = _connect(field_path)
    try:
        conn.execute(
            "UPDATE _sync_state SET last_pushed_version = ? "
            "WHERE id = 1 AND last_pushed_version < ?",
            (new_version, new_version),
        )
    finally:
        conn.close()


def push(field_path: str, central_path: str, table: str = "parcels") -> dict:
    """Full push cycle: build, apply, then advance the watermark on success."""
    delta = build_delta(field_path, table)
    if not delta["upserts"] and not delta["tombstones"]:
        return {"status": "nothing_to_push", "delta_id": delta["delta_id"]}

    result = apply_delta(central_path, delta)
    advance_watermark(field_path, delta["max_version"])
    return {"status": result, "delta_id": delta["delta_id"],
            "max_version": delta["max_version"]}


if __name__ == "__main__":
    print(push("field.gpkg", "central.gpkg"))

The ordering is the whole safety argument: the watermark advances only after apply_delta returns without raising, so a crash before that leaves the watermark untouched and the delta rebuilds identically next run.

Step-by-Step Walkthrough

1. Open connections deliberately

_connect sets isolation_level=None to take manual control of transactions and a 30-second busy timeout so short-lived locks resolve without an immediate error. WAL mode lets the push write while a reader is active.

python
conn = sqlite3.connect(path, isolation_level=None, timeout=30.0)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL;")

2. Build a minimal delta

build_delta collapses the change-log to the latest operation per feature with GROUP BY feature_id, so a feature edited ten times between syncs travels as one row. Deletes become tombstones carrying just the fid; inserts and updates carry the live attributes and geometry as GeoJSON.

python
rows = conn.execute(
    "SELECT feature_id, op, MAX(version) AS version "
    "FROM _sync_log WHERE table_name = ? AND version > ? "
    "GROUP BY feature_id",
    (table, watermark),
).fetchall()

3. Guard the apply with a ledger check

Inside the transaction, the script first asks whether this delta_id is already in _applied_deltas. If so it commits nothing and returns skipped. This check plus the ledger insert both live in the same transaction as the row changes, so idempotency and data can never diverge.

python
seen = conn.execute(
    "SELECT 1 FROM _applied_deltas WHERE delta_id = ?",
    (delta["delta_id"],),
).fetchone()
if seen:
    conn.execute("COMMIT")
    return "skipped"

4. Upsert and tombstone in one transaction

Every insert/update uses ON CONFLICT(fid) DO UPDATE, so applying a feature that already exists updates it rather than duplicating. Tombstones become DELETE statements. BEGIN IMMEDIATE acquires the write lock up front, avoiding a mid-transaction upgrade that could deadlock.

python
conn.execute(
    f"INSERT INTO {table} (fid, name, status, geom) "
    f"VALUES (?, ?, ?, GeomFromGeoJSON(?)) "
    f"ON CONFLICT(fid) DO UPDATE SET "
    f"  name = excluded.name, status = excluded.status, geom = excluded.geom",
    (f["fid"], f["name"], f["status"], f["geom"]),
)

5. Retry on lock, then advance the watermark

_with_lock_retry wraps the transaction and retries only on database is locked, backing off exponentially. When the apply finally succeeds, advance_watermark moves the device forward — and its WHERE last_pushed_version < ? guard makes even the watermark update idempotent.

python
conn.execute(
    "UPDATE _sync_state SET last_pushed_version = ? "
    "WHERE id = 1 AND last_pushed_version < ?",
    (new_version, new_version),
)

Verification

Prove the two properties that matter: the apply lands the data, and a repeat push is a no-op.

python
# Run the push twice; the second run must skip and leave counts unchanged
import sqlite3

def verify_push(field_path: str, central_path: str, table: str = "parcels") -> None:
    first = push(field_path, central_path, table)
    conn = sqlite3.connect(central_path)
    n1 = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
    ledger1 = conn.execute("SELECT COUNT(*) FROM _applied_deltas").fetchone()[0]
    conn.close()

    second = push(field_path, central_path, table)  # nothing new since watermark moved
    conn = sqlite3.connect(central_path)
    n2 = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
    conn.close()

    assert first["status"] in {"applied", "nothing_to_push"}
    assert second["status"] == "nothing_to_push", second
    assert n1 == n2, f"row count drifted on second push: {n1} -> {n2}"
    print(f"OK — {n1} features, {ledger1} deltas applied, replay was a no-op")
bash
# Structural gate on the central store after a push
sqlite3 central.gpkg "PRAGMA integrity_check;" | grep -q '^ok$' && echo "integrity OK"
sqlite3 central.gpkg "SELECT delta_id, applied_at FROM _applied_deltas ORDER BY applied_at DESC LIMIT 5;"

Alternative Approaches or Edge Cases

Applying a delta that was already committed but never acknowledged

If the apply committed but the process died before advance_watermark, the next run rebuilds the same delta (same version range, same delta_id) and the ledger check returns skipped — no duplication. The watermark then advances normally. This is the crash window the design is built to survive, and it needs no special handling.

Pushing multiple tables in one cycle

Extend build_delta to loop over a list of tables, or give each table its own delta and apply them in sequence. Keeping one delta per table makes the delta_id namespaces clean and lets a single table’s apply fail and retry without re-touching the others. If cross-table consistency matters, wrap the per-table applies in one outer transaction on the central store.

Troubleshooting

sqlite3.OperationalError: database is locked

Cause: Another writer holds the central file, or a reader under a rollback journal blocks the BEGIN IMMEDIATE.

Fix: Confirm the file is in WAL mode on every connection, keep the 30-second busy timeout, and rely on _with_lock_retry. If locks persist, a long-running reader is the culprit — shorten reader transactions. The full backoff strategy is in implementing connection retries for offline apps.

sqlite3.IntegrityError: UNIQUE constraint failed: parcels.fid

Cause: The apply used a plain INSERT instead of ON CONFLICT, or the central fid is not actually the primary key, so the upsert has no conflict target.

Fix: Ensure fid is declared PRIMARY KEY (or has a unique index) so ON CONFLICT(fid) resolves, and never fall back to bare INSERT in a retryable path.

Replayed push keeps re-applying instead of skipping

Cause: The delta_id is not stable across runs — often because it embeds a wall-clock timestamp rather than the version range.

Fix: Derive delta_id deterministically from table and the base_version-to-max_version range as shown, so the same set of changes always produces the same id and the ledger check works.