Skip to main content

The BaseLoad pattern

Every master-data entity is exposed through a Load class. They share one shape, so once you know this page, the per-object pages are mostly "what fields exist".

The master-data Loads run on Nuntiq's integration engine: SupplierLoad, OrganizationLoad, PurchaseOrderLoad, PurchaseOrderLineLoad, ReceiptLoad, CustomerDataTableLoad and UserLoad. Everything on this page applies to them. PaymentLoad and FxRateLoad have their own save rules (see their reference pages), and InvoiceLoad, IngestionLoad and LifecycleMessageLoad are not CRUD Loads at all.

Construct one​

from lib.objects.supplier import SupplierLoad

def run(context):
    load = SupplierLoad(context)

Passing context is enough. The load reads context.api and context.logger. CustomerDataTableLoad also takes the table: CustomerDataTableLoad(context, table_name='customer_data_table_1').

Source system and ownership​

Every record the engine stores belongs to a source system: the integration that wrote it. For a connector this is the connector instance's source_system configuration parameter, or the instance name when that parameter is not set.

  • Leave source_system out of your records, or set it to the same value. The framework puts it in every record and key that names none.

  • A load built with another source system, SupplierLoad(context, source_system='ZOHO_BOOKS'), puts that value in every record of save_all(), patch_all(), delete_keys() and obj.save() (also source_system=... for one call). A value other than the connector's own is accepted only under All records or All integrated records; otherwise the record is rejected (source_system_mismatch).

  • Which records the connector may change and remove is ownership_mode on the connector instance:

    • managed (default): records of its own source system; a record entered in the portal is taken over when the connector updates it.
    • own_only: records of its own source system only.
    • all_integrated: every record that came from an integration, never one entered in the portal.
    • system_of_record: every record of the resource.

    Other records are refused (owned_by_other_source). A full load (sync run) only ever removes records of the run's source system.

Keys​

Every record is found by one of its keys, tried in this order:

  1. id, the Nuntiq id;
  2. source_system + the source record id (source_record_id, or source_document_id / source_line_id / source_receipt_line_id for purchasing);
  3. the natural key, where the resource has one (supplier_number + organization_code, organization code, user login_account).

A record without a usable key is rejected (key_required). Set the source record id on everything you load: it keeps the link to your system when a number or code changes there.

Read​

# Every record, page by page under the hood
for s in load.get_all():
    print(s.supplier_number, s.supplier_name)

# Exact-match filters (each resource lists its own)
for s in load.get_all(organization_code='EMEA', is_active=True):
    ...

# Only what changed since a moment (ISO-8601)
for s in load.get_all(updated_since='2026-10-01T00:00:00Z'):
    ...

# Materialize to a list (only when you need random access)
all_suppliers = load.get_all_list()
matches = load.search(supplier_number='SUP-100')   # same as get_all_list(**filters)

# One page at a time
page = load.get(page_size=200)
while True:
    for s in page['items']:
        ...
    if not page['next_cursor']:
        break
    page = load.get(cursor=page['next_cursor'], page_size=200)

Reads page with a cursor ordered by (updated_at, id): a record changed while you read never shows up twice or gets skipped. page_size is at most 1000 (default 100). get_all() is a generator; if you break out early, you don't pay for pages you didn't read.

Filters are exact matches. A filter the resource does not have is refused (400 unknown_filter). Python booleans are sent as true and false. Deleted records are left out unless you pass include_deleted=True.

load.index_by_source('MY_ERP') returns {source_record_id: obj} for every record of that source system (suppliers, organizations, users, customer data tables).

Build a new entity​

s = load.new()                   # creates a new object, queues it for save
s.supplier_number = 'SUP-100'
s.supplier_name = 'Foo Co'
s.source_record_id = 'V-100'     # its id in your system

# Many entities have nested children. The pattern is `obj.new_<thing>()`:
loc = s.new_location()
loc.location_code = 'MAIN'
addr = loc.new_address()
addr.country_code = 'NL'

The object only exists in memory until you call save_all(). You can build up a graph, change your mind and discard it; nothing has reached Nuntiq yet.

Save: full records​

results = load.save_all()

save_all() POSTs every queued record, 1000 per request. A POST says "this is the full record":

  • the record is matched by its keys, then created or updated as a whole;
  • a field you leave out gets its default (a few fields are kept instead, such as a user's is_active or an organization's timezone; the reference pages name them);
  • a child list you send replaces your own children of that record: a child you no longer send is removed, children without a source (made in the portal) stay; a child list you leave out is left alone. An object without children of a kind leaves that list out, so its children are kept. To remove them all, assign an empty list (loc.addresses = []): an assigned list is sent also when it is empty;
  • only real changes are written. A record that matches what is stored answers unchanged;
  • a source_updated_at older than the stored one is not applied (warning stale_source_update).

save_all() returns one result per record:

{'index': 0, 'status': 'updated', 'id': '…', 'changed': ['supplier_name', 'locations'],
 'record': {...}}   # 'record' is the payload that was sent

status is created, updated, unchanged, rejected, not_found or not_applied. A rejected record carries errors (path, code, message), a stale one warnings.

Rejected records​

One bad record does not stop the others. save_all() sends every batch, then raises RecordsRejected when any record was rejected. The accepted records are saved.

from lib.objects.exceptions import RecordsRejected

try:
    load.save_all()
except RecordsRejected as e:
    for r in e.results:
        context.logger.warn(
            f"record {r['index']} not saved: {r.get('errors')}",
            detail={'record': r['record']},
        )

Pass all_or_none=True to apply a batch only when every record in it is valid. A rejection then rolls the whole batch back, and the other records of that batch answer not_applied.

Full load: a sync run​

for ext in fetch_all_suppliers():
    s = load.new()
    ...
load.save_all(full_load=True)

full_load=True makes the save a sync run:

  1. the framework starts a run (POST /v1/sync-runs);
  2. every batch is sent with the run's id;
  3. when every record was accepted, the framework completes the run, and Nuntiq soft-deletes this source system's records of the resource that the run did not touch (children go with their parent).

The table is never empty part-way, and ids stay stable. When a batch fails or any record is rejected, the framework fails the run and nothing is removed: a rejected record would otherwise be deleted. A run that saw no record cannot complete (sync_run_empty), so a broken export never wipes the data. One run per resource and source system can be open; a run left open expires after 24 hours.

Queue the complete set before calling save_all(full_load=True). Each call is its own run, so calling it on part of the data removes the rest.

Change some fields: patch_all()​

load.patch_all([
    {'supplier_number': 'SUP-100', 'organization_code': 'EMEA', 'is_active': False},
    {'source_record_id': 'V-200', 'supplier_name': 'Bar Co'},
])

A PATCH merges: only the fields you send change, None clears a field, everything else stays. It never creates a record (an unknown key answers not_found). Every item needs a key. Child lists merge by their keys; a child with '_delete': True is removed:

load.patch_all([{
    'source_record_id': 'V-100',
    'locations': [{'location_code': 'OLD', '_delete': True}],
}])

You can also pass entity objects; only their changed fields are sent. patch_all() returns the per-record results and raises RecordsRejected like save_all().

Update one entity​

for s in load.get_all(organization_code='EMEA'):
    if s.supplier_name.startswith('OLD '):
        s.supplier_name = s.supplier_name[4:]
        s.save()                 # POSTs this one record, as a whole

obj.save() sends the whole object, like save_all(). A record you read and save back keeps every field you did not touch, because it was read whole. An object you built with new() carries only what you set, so the fields you leave out go back to their defaults. Use patch_all() when you want to change a few fields without reading first.

To switch a record off, call obj.deactivate(). It PATCHes is_active (organizations: active) to false and changes nothing else. Purchase orders, PO lines and receipts have no active flag: deactivate() raises ValueError there; use delete().

Delete​

load.delete_keys([
    {'source_record_id': 'V-100'},
    {'supplier_number': 'SUP-200', 'organization_code': 'EMEA'},
    {'id': '0b6f…'},
])

obj.delete() deletes one object by its Nuntiq id. Objects you read keep that id (suppliers in supplier_id, supplier locations in supplier_location_id, the others in id), and so do objects after a save. delete_keys() works with any key and is the usual choice.

Deletes are soft (is_deleted, deleted_at) and by key only. They touch only your own source system's records (any record in system-of-record mode). Sending a deleted record again restores it. Deleting a purchase order also deletes its lines and receipts; deleting a line its receipts. An organization unit with live children cannot be deleted.

There is no delete by filter. To remove what your source no longer has, use a full load. delete_where() raises NotImplementedError on these Loads.

Retries and idempotency​

Every engine write (save_all(), patch_all(), delete_keys(), obj.save(), obj.delete(), obj.deactivate()) carries an Idempotency-Key. A connection error, a 502, 503 or 504, or a 409 idempotency_in_progress is retried up to 3 times (after 1, 3 and 9 seconds) with the same key. A write the API already applied is then answered from the replay, not applied twice. Other errors raise ApiError.

Validation​

Before anything is sent, save_all() and obj.save() check required fields and types. A missing required field raises ValidationError and nothing is sent. The engine checks the rest per record (lengths, values, references) and answers rejected for that record only.

from lib.objects.exceptions import ValidationError

try:
    load.save_all()
except ValidationError as e:
    context.logger.warn(f"validation failed: {e.field_errors}")

Order of operations​

Records reference each other by code. Save the referenced ones first: organizations before suppliers, purchase orders before their lines, lines before receipts. A PO line whose source_document_id names no purchase order of your source system is rejected.

org_load = OrganizationLoad(context)
sup_load = SupplierLoad(context)

# 1. Organizations first
for ext in fetch_external_orgs():
    o = org_load.new()
    o.code = ext['code']
    o.name = ext['name']
    o.organization_type = 2
    o.source_record_id = ext['id']
org_load.save_all()

# 2. Then the suppliers that reference them
for ext in fetch_external_suppliers():
    s = sup_load.new()
    s.supplier_number = ext['number']
    s.supplier_name = ext['name']
    s.organization_code = ext['org_code']  # references organization.code
    s.source_record_id = ext['id']
sup_load.save_all()

The exception: InvoiceLoad​

Invoices are produced by the platform, not created by connectors. So InvoiceLoad doesn't follow the new() / save_all() pattern. It exposes claim() / get_by_token() / acknowledge() / integration_result(). See Invoice claim flow.

What's next​