Syncing — snapshot & deltas

A producer pushes data two ways: a full snapshot to establish (and periodically re-establish) the baseline, and incremental deltas for the changes in between. Both are driven by the entities you build.

Full snapshot

client.snapshot(entities) streams every entity as gzipped NDJSON and runs the three-step upload (init → upload → finalize) for you. Pass any iterable so the catalog never lands in memory.

from voicebot import client
from monoverse_voicebot import product, category

def catalog():
    yield category(external_id="shop:category:phones", name="Phones", slug="phones")
    for row in db.stream_products():
        yield product(
            external_id=f"shop:product:{row.id}",
            name=row.title,
            price_amount=round(row.price * 100),
            currency="UAH",
            stock_status="instock" if row.in_stock else "outofstock",
            categories=row.category_slugs,
        )

result = client.snapshot(catalog())
print(f"snapshot {result.sync_id}: {result.total} records", result.counts)

The returned SnapshotResult has sync_id, total, and per-kind counts.

A full snapshot is the source of truth — an empty one is refused

The backend tombstones anything not present in a full snapshot, so an accidental empty snapshot would wipe your catalog. client.snapshot() refuses to push zero entities and raises a ConfigError instead. If you only need to change a few items, use send_events — never an empty/partial snapshot.

The snapshot file is capped at 200 MB (snapshot_max_bytes); over that the client raises before uploading. Sync in smaller scopes if you hit it.

Incremental deltas

After the baseline, push only what changed. Wrap entities into operations and call send_events. An upsert replaces the whole entity — a field you leave out does not keep its old value — so build it with every field your snapshot sends, not only the one that changed:

from voicebot import client
from monoverse_voicebot import product, to_upsert_operation, delete_operation

result = client.send_events([
    to_upsert_operation(
        product(
            external_id="shop:product:iphone-15-pro",
            name="iPhone 15 Pro",
            price_amount=52999,  # the new price
            currency="UAH",
            stock_status="instock",
            categories=["phones"],
        ),
    ),
    delete_operation("product", "shop:product:discontinued-sku"),
])

print(f"processed {result.processed}, errors {result.error_count}")

The client splits operations into wire batches automatically — capped at 500 ops and 5 MB per batch. Each batch carries a fresh batch_id that doubles as its Idempotency-Key, so a resign-and-resend of the same batch is not applied twice: if the first attempt was applied, the resend returns its original result. After a 429, a 5xx or a dropped connection the client resends a batch up to 3 times by default, re-signed each time (Operations → Retry semantics). If a batch that carries upserts comes back with idempotent_replay=True, processed 0 and no errors, an earlier attempt still holds it or never applied it (Errors → Idempotency): the client asks again after 1, 2 and 4 s, and if the answer stays the same it reports the batch as unconfirmed instead of done.

Call it from a worker

A single send_events call can block for minutes — per batch, up to 3 waits of 60 s on 429s plus up to 7 s of re-checks. Call it from a worker, a cron job, a Celery task or CatalogScheduler, never inside a web request handler.

Reading the result

EventsResult gives you processed, error_count, unconfirmed and the per-batch detail in batches: each EventsBatchResult has batch_id, processed, idempotent_replay, errors and confirmed. Per-op failures (a missing required field, money sent as a float) come back as typed errors — they do not raise; inspect them:

for batch in result.batches:
    for err in batch.errors:
        print(f"op {err.index} {err.entity_kind}:{err.external_id} → {err.error}")

Unconfirmed batches

A batch is unconfirmed (batch.confirmed is False, and it counts in result.unconfirmed) when it carries at least one upsert and the server still answers processed 0, no errors and idempotent_replay=True after the client asked again at 1, 2 and 4 s. An earlier attempt still holds that batch, or never applied it. Its processed stays 0: it is never counted as applied.

Do not advance your change watermark for that run, and on the next run send the current state of those entities again (each call mints a new batch_id):

result = client.send_events(ops)
if result.unconfirmed == 0:
    save_watermark(run_started_at)  # otherwise the next run resends these entities

A delete-only batch is never reported unconfirmed — resending a delete is harmless.

When a batch raises

Batches go out in order, and send_events stops at the first one that raises. The error carries the results of the batches sent before it as partial_result — an EventsResult declared on VoiceBotError, whose batches is empty if the first batch failed. The failing batch's outcome is unknown, and the batches after it were not sent. This covers TransientError (after the resends), ConfigError (a 4xx), ProtocolVersionError (426) and NotPairedError.

from monoverse_voicebot import VoiceBotError

try:
    client.send_events(ops)
except VoiceBotError as exc:
    landed = exc.partial_result.processed if exc.partial_result else 0
    log.warning("%s ops landed before the failure", landed)
    raise

Snapshot vs deltas — when to use which

	client.snapshot()	client.send_events()
What it sends	A complete baseline of every kind	Only the upserts/deletes you pass
Transport	gzip-NDJSON file upload	/events batches (≤500 ops / ≤5 MB)
Frequency	First setup, then periodically (e.g. nightly)	Often (minutes), as data changes
Deletes	Reconciles hard deletes via server tombstones	Only the deletes you send explicitly
Cost	Proportional to catalog size	Tiny — proportional to churn

Why a periodic full

Deltas carry only the changes you observe and send. A periodic full snapshot is the safety net for:

• Hard deletes you can't detect as a delta — the snapshot's server-side tombstone pass removes anything no longer present.
• Out-of-band edits that bypass your change tracking — the full re-syncs everything regardless.

Scheduling

Don't hand-code the loop — CatalogScheduler runs frequent deltas plus a periodic full in background threads, with a without-overlapping guard (a slow run never starts a second copy) and errors routed to on_error so the daemon stays alive:

import logging
from monoverse_voicebot import VoiceBotClient, CatalogScheduler

client = VoiceBotClient(...)

scheduler = CatalogScheduler(
    delta=lambda: client.send_events(changed_ops()),   # default cadence: every 15 min
    full=lambda: client.snapshot(all_entities()),      # default cadence: every 24 h
    run_full_on_start=True,
    on_error=lambda phase, exc: logging.getLogger(__name__).exception("%s failed", phase),
)
scheduler.start()
# … keep the process alive; call scheduler.stop() on shutdown

Run it as a long-lived process under a supervisor (systemd, Docker restart: unless-stopped) and call scheduler.stop() on shutdown. Tune delta_interval_s / full_interval_s to your catalog's churn.

Next

• Operations — status heartbeat, unpair, retry semantics, and running on a schedule.