Sync engine

Python SDK

aiotic.sync sends master data to AIOTIC only when something changed, in the right order (per record and across kinds), in parallel across records, and reports what happened.

Concepts

Type	Role
ChangeEvent	one change in AIOTIC terms: kind (customer / product / mapping), op (upsert / delete), key, data. Constructors: customer_upsert(number, **fields), customer_delete(number), product_upsert(item_number, language_code, description=…, remark=None), product_delete(…), mapping_upsert(customer_number, customer_item_number, item_number=…, language_code=…), mapping_delete(…)
StateStore	remembers the fingerprint last sent per key (+ watermarks). InMemoryStateStore, HashStateStore(path) (SQLite), or your own
SyncEngine(client, state, concurrency=4, dry_run=False)	applies events; skips unchanged; FK ordering; parallelism; report
SyncReport	sent, deleted, skipped_unchanged, failed, errors[], duration
PollingChangeSource(name, fetch_changed_since, engine, interval)	watermark-based poller for ERPs without events

API

engine = SyncEngine(client, state=HashStateStore("sync-state.db"), concurrency=6)

engine.apply(event, force=False) -> bool            # one event; True when a request was made
engine.apply_many(events, force=False) -> SyncReport # batch: one record at a time in the order given; a mapping after its parents, a parent delete after its mappings; otherwise customers→products→mappings for upserts, reversed for deletes
engine.reconcile(customers=…, products=…, mappings=…, delete_missing=True, allow_empty=False) -> SyncReport
engine.bootstrap_state_from_aiotic() -> int         # seed fingerprints from what AIOTIC holds

• Unchanged fingerprint → skipped, no request.
• 404 on an upsert → reported as "referenced customer/product missing (upsert those first)".
• 404 on a delete → counted as deleted (already gone).
• Other API errors → failed with the message; the fingerprint is not updated, so the next run retries.
• dry_run=True logs what would be sent.

Mapping ERP records

Keep the mapping in one function per kind so events and reconciliation share it:

def customer_event(c) -> ChangeEvent:
    if c.blocked:
        return ChangeEvent.customer_delete(c.no)
    return ChangeEvent.customer_upsert(c.no, name=c.name, address=c.address, postal_code=c.post_code, city=c.city,
                                       vat_number=c.vat_registration_no, email=c.email, phone_number=c.phone,
                                       contact_person=c.contact, coc_number=c.registration_no, home_page=c.home_page)

Fields you do not have are simply omitted (null). Empty strings become null in the fingerprint, so a value that flips between "" and None does not cause churn.

Event sources

# webhook / outbox → generic JSON → engine
from aiotic.webhooks import parse_change_events
engine.apply_many(parse_change_events(payload))

# polling
src = PollingChangeSource("customers", fetch_changed_since, engine, interval=300)
threading.Thread(target=src.run_forever, daemon=True).start()

The watermark advances only after a batch with zero failures.

Reconciliation

reconcile() expects the complete current data set per kind you pass (generators are fine — 200 000 products stream through). It sends differences and, with delete_missing=True, deletes keys it sent earlier that are no longer present. Pass only the kinds you have complete data for.

An empty data set for a kind that has records in the state store is refused: nothing is deleted and the report carries one failure saying so, because a broken export must not empty a tenant. Pass allow_empty=True (CLI: --allow-empty) when the data set really is empty and the deletions are intended.

Before the first reconcile on a tenant that already has data: engine.bootstrap_state_from_aiotic() (or aiotic sync bootstrap-state). Products that AIOTIC returns without a description are not seeded (an upsert needs one); the next reconcile sends them once with yours.

Concurrency and keys

Use 4–8 workers. Events for the same record are applied one after another, in the order you pass them, never concurrently; a mapping waits for the customer and product events that precede it, a customer or product delete waits for the mapping events that precede it, and everything else runs in parallel. A batch that creates and then removes a customer, a product and their mapping therefore succeeds in source order. Configure AIOTIC_SYNC_API_KEY; the client routes master-data calls to it automatically. The client's rate limiter is shared by all workers.

Observability

Log the SyncReport per run; alert on failed > 0 for two consecutive runs. Keep errors[] — they name the record and the reason.