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:

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),
    ),
    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 replays server-side (idempotent_replay=True) rather than double-applying.

Reading the result

EventsResult gives you processed, error_count, and per-batch detail. 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}")

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.