Skip to main content

The BaseLoad pattern

Every entity (Supplier, Organization, PurchaseOrder, Receipt, Payment, etc.) is exposed through a Load class. They all share the same shape, so once you know this page, the per-object pages are mostly just "what fields exist".

Construct one​

from lib.objects.supplier import SupplierLoad

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

Passing context is enough — the load reads context.api to do its work.

Read​

# Iterate every row, paginated under the hood
for s in load.get_all():
    print(s.supplier_number, s.supplier_name)

# Materialize to a list (only when you genuinely need random access)
all_suppliers = load.get_all_list()

# Search by field — kwargs become query params
results = load.search(supplier_name='Acme')

# One page at a time, with manual paging
page1 = load.get(page=1, page_size=50)
print(page1['count'], 'total rows;', len(page1['items']), 'on this page')

get_all() is a generator — it fetches pages lazily as you iterate. If you break out early, you don't pay for pages you didn't read.

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.is_active = True

# 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 complex graph, change your mind, and discard it — nothing has hit Nuntiq yet.

Save​

load.save_all()                  # POST everything queued. Default behavior:
                                 # upsert on natural keys, leave other rows alone.

load.save_all(full_load=True)    # WIPES the existing table for this customer
                                 # and replaces it with what's queued. Only use
                                 # this when you're authoritative.

save_all() chunks at 1000 rows per request automatically — you can queue 10,000 entities and it'll send 10 batches.

warning

full_load=True is destructive. Don't use it unless your connector is the only source of truth for that table for the customer.

Update an existing entity​

for s in load.get_all():
    if s.supplier_name.startswith('OLD '):
        s.supplier_name = s.supplier_name[4:]
        s.save()                 # POST just this one back

obj.save() only sends the fields you actually changed (dirty tracking). Setting a field to its current value is a no-op.

Delete or deactivate​

There are two ways. Both go through the bulk DELETE endpoint. Pick the one that fits.

One row by id (you have the entity in hand):

s = next(load.search(supplier_number='SUP-100'))
s.deactivate()                   # soft: is_active = false   (supplier/org/payment/CDT only)
# or
s.delete()                       # soft: is_deleted = true   (everything)

Many rows by filter (you don't need to fetch first):

load.delete_where(
    parameters=[
        {'updated_before_date': '2025-01-01'},  # one filter row
        {'supplier_number': 'SUP-X'},           # OR another filter row
    ],
    action='delete',                            # or 'deactivate' (where supported)
)

Each dict in parameters is one filter; multiple dicts run as separate UPDATEs. Common keys: <id_field>, <natural_key>, organization_code, created_before_date, created_after_date, updated_before_date, updated_after_date. Exact accepted keys depend on the resource — check the Customer API reference for the DELETE endpoint of the resource if you need something exotic.

:::note Soft-delete only on PO / PO-line / receipt PurchaseOrderLoad, PurchaseOrderLineLoad, and ReceiptLoad only support soft-delete. They have no is_active column and the DELETE endpoint does not accept an action field. Calling .deactivate() or delete_where(..., action='deactivate') on them raises ValueError — call .delete() or delete_where(...) (which defaults to soft-delete) instead. Soft-deleting a PO header cascades to its lines and receipts; soft-deleting a line cascades to its receipts. :::

Validation​

Required-field checks run at save_all() time. If a required field is missing on any queued object, the whole batch is rejected with a ValidationError and no rows are saved. Catch it if you need partial success:

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 matters​

If your connector creates organizations and suppliers that reference those organizations, save organizations first. The API enforces referential integrity — supplier rows pointing at a missing organization_code will be rejected.

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

# 1. Build + save orgs first
for ext in fetch_external_orgs():
    o = org_load.new()
    o.code = ext['code']
    o.name = ext['name']
    o.organization_type = 2
org_load.save_all()

# 2. Now safe to 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
sup_load.save_all()

The exception: InvoiceLoad​

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

What's next​

  • Delta state — incremental sync with a high-water cursor.
  • Logging — surfacing what your save_all calls actually did.
  • Entity reference — field-by-field for each Load class.