CustomerDataTableLoad

Load class for the customer data tables: reference data for rules and enrichment (cost centres, GL accounts, approval matrices). Each tenant has twenty tables, customer_data_table_1 to customer_data_table_20, with the same generic columns. What each column holds is agreed per tenant; the rules and enrichment that read the table use the same columns.

Import

from lib.objects.customer_data_table import CustomerDataTableLoad

Constructor takes a table name

load = CustomerDataTableLoad(context, table_name='customer_data_table_1')
# Hits /v1/customerdatatable/customer_data_table_1

Any other name is refused (404 unknown_table).

Only the custom entrypoints (get_custom_1..N) may read and write customer data tables; other entrypoint keys get a 403.

Standard BaseLoad methods

get_all(), search(), new(), save_all(), patch_all(), delete_keys(), index_by_source(), obj.save(), obj.delete(), obj.deactivate(). See BaseLoad.

Fields

Use .set() and .get() for the columns:

Field	Notes
source_record_id	The record's id in your system. With the source system, the key
text_field_1..30	Text, up to 4000 characters
number_field_1..10	Numbers
date_field_1..10	ISO dates
is_active	Default true
source_system	Leave out, or the connector's source system
source_updated_at	When your system last changed the record; an older value than the stored one is not applied

Read-only on GET: id, created_at, updated_at. A field that is not a column is refused (unknown_field).

rec = load.new()
rec.set('source_record_id', 'CC-001')
rec.set('text_field_1', 'EMEA Marketing')
rec.set('text_field_2', 'Active')
load.save_all()

for rec in load.get_all():
    print(rec.get('source_record_id'), rec.get('text_field_1'))

Keys

A record is found by its Nuntiq id or by source_system + source_record_id. The same source_record_id from two source systems is two records.

Full load

save_all(full_load=True) is a sync run: after the last batch, your records of this table that the run did not send are soft-deleted. When any record was rejected, nothing is removed.

Example: import cost centres from the ERP

from lib.objects.customer_data_table import CustomerDataTableLoad

def run(context):
    load = CustomerDataTableLoad(context, table_name='customer_data_table_1')

    erp_cost_centers = fetch_cost_centers_from_erp()  # returns dicts

    for cc in erp_cost_centers:
        rec = load.new()
        rec.set('source_record_id', cc['id'])
        rec.set('text_field_1', cc['name'])
        rec.set('text_field_2', cc['status'])
        rec.set('text_field_3', cc['owner_email'])

    load.save_all(full_load=True)     # the complete list: removes cost centres the ERP dropped
    return {'imported': len(erp_cost_centers)}

Search

Exact-match filters: is_active, text_field_1, number_field_1, date_field_1, source_system, source_record_id, plus updated_since and include_deleted.

for rec in load.get_all(text_field_1='EMEA Marketing'):
    ...

Delete and deactivate

# Soft delete by key
load.delete_keys([{'source_record_id': 'CC-OLD'}])

# Switch a record off and keep it
load.patch_all([{'source_record_id': 'CC-OLD', 'is_active': False}])

# The same for a record you read: deactivate() PATCHes is_active to false
for rec in load.search(source_record_id='CC-OLD'):
    rec.deactivate()

There is no delete by filter: delete_where() raises NotImplementedError.