Syncing QGIS Edits to Cloud Annotation Platforms

Syncing QGIS edits to a cloud annotation platform requires a deterministic extract-transform-upload pipeline built in PyQGIS. The correct approach reads only the pending changes from QgsVectorLayerEditBuffer, reprojects all geometries to EPSG:4326 (WGS 84), maps QGIS field names to the platform’s JSON schema, and pushes authenticated batch payloads with per-request idempotency keys and exponential backoff. This eliminates manual shapefile handoffs, prevents silent geometry corruption, and keeps desktop GIS work traceable within the human-in-the-loop validation cycle.

# Why This Matters in Geospatial Annotation Pipelines

Cloud annotation APIs never accept native .qgz or .shp files — they expect structured GeoJSON payloads with explicit coordinate arrays, label dictionaries, and metadata tags. When teams skip a formal sync pipeline and export manually, three failure modes appear consistently: mixed coordinate reference systems across batches cause geometry offsets that collapse IoU scores; NaN float attributes silently break strict JSON parsers and drop entire feature batches; and duplicate uploads from manual re-runs corrupt the annotation record without warning. A scripted, idempotent pipeline closes all three gaps and feeds directly into the dataset versioning audit trail.

# Pipeline Architecture

The three-phase pipeline below maps directly to the extract, transform, and upload operations that every production sync must implement.

Skipping any phase introduces silent failures. The transform step is especially critical: most ML annotation engines reject features with unclosed polygon rings or coordinates outside [-180, 180] / [-90, 90], and the rejection often surfaces as a non-specific 422 Unprocessable Entity with no per-feature detail.

# Step-by-Step Implementation

# Step 1 — Extract Pending Edits from the Edit Buffer

Query only modified and added features from the active editing session. Reading the full layer generates redundant payloads and risks overwriting annotations that other team members have already reviewed during the validation cycle.

from qgis.core import QgsProject

LAYER_NAME = "annotation_edits"

def get_pending_feature_ids(layer_name: str) -> list[int]:
    layers = QgsProject.instance().mapLayersByName(layer_name)
    if not layers:
        raise RuntimeError(f"Layer '{layer_name}' not found in project.")
    layer = layers[0]

    if not layer.isEditable():
        raise RuntimeError(f"Layer '{layer_name}' must be in edit mode.")

    edit_buffer = layer.editBuffer()
    changed_ids = list(edit_buffer.changedAttributeValues().keys())
    added_ids   = list(edit_buffer.addedFeatures().keys())
    return list(set(changed_ids + added_ids))

QgsVectorLayerEditBuffer tracks three separate dictionaries: changedAttributeValues, changedGeometries, and addedFeatures. Unioning the first and last captures both attribute-only updates and new polygon draws; changedGeometries is implicitly included because geometry edits also update attributes in most annotation workflows.

# Step 2 — Reproject and Validate Geometries

Transform all coordinates to EPSG:4326 before serialising to GeoJSON. Cloud annotation APIs enforce the GeoJSON specification’s WGS 84 requirement; submitting features in a projected CRS such as EPSG:32633 causes silent coordinate drift that can shift polygon vertices by hundreds of metres on the platform map. This CRS contract is the same one described in the broader guide to coordinate reference systems in annotation pipelines.

import json
from qgis.core import (
    QgsCoordinateTransform,
    QgsCoordinateReferenceSystem,
    QgsGeometryValidator,
    QgsProject,
)

def reproject_and_validate(geom, source_crs) -> dict | None:
    """Reproject geometry to EPSG:4326, validate, and return as GeoJSON dict."""
    if not geom or geom.isEmpty():
        return None

    target_crs = QgsCoordinateReferenceSystem("EPSG:4326")
    xform = QgsCoordinateTransform(source_crs, target_crs, QgsProject.instance())
    geom.transform(xform)

    errors: list = []
    QgsGeometryValidator.validateGeometry(geom, errors)
    if errors:
        geom = geom.makeValid()  # attempt auto-repair
        if geom is None or geom.isEmpty():
            return None  # discard unrecoverable geometry

    return json.loads(geom.asJson())

QgsGeometryValidator catches self-intersecting rings and unclosed polygons before they hit the network. makeValid() resolves most issues by splitting self-intersections into separate parts — confirm the output geometry type matches your annotation schema (MultiPolygon where Polygon is expected may require an additional unwrap step).

# Step 3 — Map Attributes to the Platform Schema

Flatten QGIS field values, sanitise floats, and rename fields to match the target annotation API’s expected property names. Mismatched field names cause silent attribute drops; NaN floats break JSON serialisation in the strict parsers used by Label Studio and similar platforms.

import math

# Field mapping: QGIS field name -> platform schema name
FIELD_MAP = {
    "label_class": "category",
    "annotator_id": "reviewer",
    "confidence": "score",
    "review_note": "comment",
}

def sanitize_value(v):
    """Return None for NULL or NaN; pass everything else through."""
    if v is None:
        return None
    if isinstance(v, float) and math.isnan(v):
        return None
    return v

def map_attributes(feature, layer_fields) -> dict:
    raw = {
        field.name(): sanitize_value(feature[field.name()])
        for field in layer_fields
    }
    return {FIELD_MAP.get(k, k): v for k, v in raw.items()}

Define FIELD_MAP per project and confirm the platform schema against its OpenAPI or JSON Schema spec — check whether the platform treats unlisted extra properties as errors or silently ignores them before deciding how aggressively to filter.

# Step 4 — Batch Upload with Idempotency and Retry

Upload in configurable batches with per-batch Idempotency-Key headers and exponential backoff. Idempotency keys prevent duplicate ingestion when network retries fire; without them, a single transient 502 can result in the same feature appearing twice in the annotation record and corrupting confidence scores aggregated at the platform level.

import time
import uuid
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

TARGET_API_URL = "https://api.your-platform.com/v1/annotations/batch"
AUTH_TOKEN     = "YOUR_API_KEY"   # use QgsAuthManager in production (see below)
BATCH_SIZE     = 50

def build_session() -> requests.Session:
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=1.5,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"],
    )
    session.mount("https://", HTTPAdapter(max_retries=retry))
    session.headers.update({
        "Authorization": f"Bearer {AUTH_TOKEN}",
        "Content-Type": "application/json",
    })
    return session

def upload_batch(session: requests.Session, records: list[dict]) -> None:
    for i in range(0, len(records), BATCH_SIZE):
        chunk = records[i : i + BATCH_SIZE]
        idempotency_key = str(uuid.uuid4())   # one key per batch, reused on retries
        headers = {"Idempotency-Key": idempotency_key}
        resp = session.post(
            TARGET_API_URL,
            json={"features": chunk},
            headers=headers,
            timeout=30,
        )
        resp.raise_for_status()

        # Respect rate-limit headers if the platform exposes them
        remaining = resp.headers.get("X-RateLimit-Remaining")
        if remaining and int(remaining) < 5:
            time.sleep(2)

        print(f"Uploaded {len(chunk)} features — status {resp.status_code}")

# Step 5 — Orchestrate the Full Sync

def sync_qgis_edits_to_cloud(layer_name: str = LAYER_NAME) -> None:
    layers = QgsProject.instance().mapLayersByName(layer_name)
    layer  = layers[0]
    source_crs = layer.crs()
    pending_ids = get_pending_feature_ids(layer_name)

    if not pending_ids:
        print("No pending edits to sync.")
        return

    records: list[dict] = []
    skipped = 0

    for fid in pending_ids:
        feature = layer.getFeature(fid)
        if not feature.isValid():
            skipped += 1
            continue

        geom_dict = reproject_and_validate(feature.geometry(), source_crs)
        if geom_dict is None:
            skipped += 1
            continue

        records.append({
            "id":             str(fid),
            "geometry":       geom_dict,
            "properties":     map_attributes(feature, layer.fields()),
            "sync_timestamp": time.time(),
        })

    print(f"Syncing {len(records)} features ({skipped} skipped).")
    upload_batch(build_session(), records)
    print("Sync complete.")

Connect to the committedChanges signal so the sync fires automatically after every QGIS commit without manual intervention:

layer.committedChanges.connect(lambda: sync_qgis_edits_to_cloud(LAYER_NAME))

# Spatial Parameters and Configuration Reference

# Common Errors and Fixes

RuntimeError: Layer 'annotation_edits' must be in edit mode : Cause: the editing session was committed or rolled back before sync_qgis_edits_to_cloud() was called, clearing the buffer. : Fix: connect the sync function to layer.committedChanges so it fires immediately after the QGIS commit, before the buffer is discarded.

422 Unprocessable Entity on batch upload : Cause: the platform’s schema validator rejected at least one feature — commonly an unclosed polygon ring, a coordinate pair outside [-180, 180] / [-90, 90], or a required property missing after field mapping. : Fix: shrink BATCH_SIZE to 1 to isolate the offending feature, inspect the response body for field-level error detail, and add the missing field to FIELD_MAP or tighten the reproject_and_validate gate.

Silent duplicate annotations after a network retry : Cause: Idempotency-Key was regenerated per retry attempt rather than per batch, so the platform treated each retry as a distinct write. : Fix: generate one UUID per batch before entering the retry loop and reuse it on all attempts for that batch.

ValueError: Out of range float values during JSON serialisation : Cause: sanitize_value was not applied and a QGIS field contained float('nan') or float('inf'). : Fix: confirm map_attributes calls sanitize_value on every field value, including any nested dictionary values if QGIS fields store JSON strings that are parsed and re-serialised.

# Credential Management with QgsAuthManager

Never hardcode API tokens in .py files or embed them in .qgz project files shared across a team. Use QGIS Authentication Manager (QgsAuthManager) to store credentials under a named configuration ID and retrieve them at runtime:

from qgis.core import QgsApplication

auth_mgr = QgsApplication.authManager()
# Retrieve config by the ID set in QGIS > Settings > Authentication
cfg = {k: v for k, v in auth_mgr.availableAuthMethodConfigs().items()
       if v.name() == "cloud_annotation_api"}

For delta recovery — resuming a failed mid-batch sync without re-uploading already-accepted features — maintain a local SQLite log of synced feature IDs alongside their sync_timestamp. On restart, exclude IDs present in the log from pending_ids before calling upload_batch. This log also feeds the SHA hashing audit trail that version-stamps each feature payload for rollback and change tracking.

This integration is one component of the broader Human-in-the-Loop Validation Cycles workflow, which covers the full reviewer feedback loop from pre-labeling through quality gates.

Related

• Human-in-the-Loop Validation Cycles — parent page covering the full reviewer feedback loop
• QGIS Plugin Ecosystem for Annotation Teams — toolchain context: which QGIS plugins complement a PyQGIS sync script
• Coordinate Reference Systems in Annotation Pipelines — the CRS contracts that govern the EPSG:4326 requirement in this pipeline
• Tracking Annotation Changes with SHA Hashing — version-stamp each feature payload before upload for audit and rollback
• Labeling Workflows & Toolchain Integration — the top-level section covering the full annotation toolchain