Sync engine

Python SDK

aiotic.sync sends master data to AIOTIC only when something changed, in the right order, in parallel, 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: upserts customers→products→mappings, deletes reversed
engine.reconcile(customers=…, products=…, mappings=…, delete_missing=True) -> 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.

Before the first reconcile on a tenant that already has data: engine.bootstrap_state_from_aiotic() (or aiotic sync bootstrap-state).

Concurrency and keys

Use 4–8 workers. 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.