Skip to main content

REST API pull

The most common connector: read records that changed in the customer's system since the last run and write them into Nuntiq. This page builds get_supplier for an ERP with a REST API. The erp_rest_* templates contain this pattern for every master-data entrypoint; start there for a real connector.

The scenario​

The ERP lists vendors at GET /api/v1/vendors?modifiedSince=...&page=N, 100 per page, oldest change first, with an API key in the X-API-Key header. Each vendor has an id, a number, a name, an address and bank accounts. The ERP answers 429 when called too often.

Settings and secrets​

NameWhereExample
connector_namesettingacme_erp (names the delta cursor; unique per connector)
source_systemsettingACME_ERP (the identity written on the suppliers)
base_urlsettinghttps://erp.example.com/api/v1
lookback_dayssetting30 (first run only)
api_keysecretthe ERP's API key

The connector​

"""get_supplier: ERP vendors -> Nuntiq suppliers (changes since the last run)."""
from lib.helpers import Settings, RunSummary, HttpClient, ApiKeyAuth, incremental, country_code, utc_iso
from lib.objects.supplier import SupplierLoad


def run(context):
    s = Settings(context)
    summary = RunSummary(context)
    erp = HttpClient(s.str('base_url'), auth=ApiKeyAuth(s.secret('api_key')), context=context, name='ERP')
    load = SupplierLoad(context, source_system=s.source_system)

    with incremental(context, 'suppliers') as run:
        for v in erp.paginate('vendors', params={'modifiedSince': run.since}, items='data',
                              style='page', page_size=100):
            sup = load.new()
            sup.source_record_id = str(v['id'])
            sup.supplier_number = v['number']
            sup.supplier_name = v['name']
            loc = sup.new_location()
            loc.location_code = 'MAIN'
            addr = loc.new_address()
            addr.street = v.get('street')
            addr.city = v.get('city')
            addr.country_code = country_code(v.get('country'))
            run.seen(utc_iso(v['modifiedAt']))
            summary.add('suppliers')
        load.save_all()

    return summary.result(cursor=run.new_cursor)

What the helpers do for you​

  • Settings: s.str('base_url') fails the run with "base_url is required" when it is empty; s.secret('api_key') the same for the secret.
  • Retries: a 429 waits as long as the ERP's Retry-After says and tries again; 502, 503, 504 and dropped connections are retried with growing waits. A GET is always safe to repeat.
  • Paging: paginate() follows page=1,2,3 until a short or empty page. Other APIs use style='offset', 'cursor', 'next_url' (for example items='value', next_url='@odata.nextLink' for Microsoft APIs) or 'link_header'.
  • The cursor: run.since is where this run starts (the job parameter since_override, else the cursor of the last successful run, else now minus lookback_days). The next run starts at the highest value passed to run.seen(). When the run fails, the cursor stays where it was and the next run reads the same changes again. Saving is idempotent, so that is safe.
  • Big backlogs: incremental(context, 'suppliers', max_records=5000) and check run.full in the loop; the next run continues where this one stopped. The source must list oldest first for that.

A login other than an API key​

Only the auth= changes. For OAuth 2 client credentials:

auth = OAuth2ClientCredentials(s.str('token_url'), s.secret('client_id'), s.secret('client_secret'),
                               scope=s.str('scope', None), cache=TokenCache(context))

Every login method is in Helpers and in its own template. See also OAuth 2.0.

Testing it locally​

Put the settings and secrets in a settings file of the toolkit, point base_url at the ERP's sandbox and run python nuntiq.py <settings> get_supplier. The suppliers land in the toolkit's fake Nuntiq (fixtures/supplier.json); nothing reaches a real tenant.

Deployment notes​

  • The ERP must accept calls from Nuntiq's outbound IP address on port 443.
  • Schedule get_supplier with a scheduler task of type connector_dispatch. For a full reload, set the job parameter since_override on a one-off run.
  • Give every record its id in the source system (source_record_id): the next run updates the same supplier instead of creating a second one.