Helpers

lib.helpers holds the building blocks every connector template is made of. Use them in your own code as well: they take care of retries, paging, token handling, file claiming and the push contract, so an entrypoint holds only what is specific to the customer's system.

from lib.helpers import Settings, HttpClient, OAuth2ClientCredentials, incremental, run_push

They never break your code. A customer connector keeps the code it was created with while the framework is upgraded underneath it. Everything exported from lib.helpers is therefore permanent: later versions add classes, functions and optional arguments, but never remove or rename them or change what they do. The docstrings in lib/helpers/*.py (in the toolkit) are the full reference; this page is the map.

Settings and results

Helper	What it does
`Settings(context)`	Reads settings and secrets with a type: `.str`, `.int(min=, max=)`, `.float`, `.bool`, `.json`, `.list`, `.choice(key, options)`, `.date`, `.secret`, `.pem`. A key without a default is required. A missing or malformed value raises `ConfigError` naming the key and where it is set. `.connector_name`, `.source_system`, `.param(key)` (job parameters).
`RunSummary(context)`	Counters (`.add('created')`), `.warn()`, `.error()`; `.result()` is the dict to return from `run()`. With errors it reports `success: false`.
`ConfigError`	A setting or secret is wrong. The run fails with that message.
`RetryableError`	The target is down or busy: try again later. Claimed invoices go back, files stay.
`PermanentError(code, message)`	The target refused this record: do not retry. `code` becomes the failure code.
`TargetError`	Raised by `HttpClient` and the file stores; `.retryable` tells which of the two above it is.

Settings arrive from the admin portal as text ("false", "30"). Settings converts them, so if s.bool('attach_pdf', True): behaves the same in the portal and in the toolkit.

s = Settings(context)
base_url = s.str('base_url')                         # required
page_size = s.int('page_size', 100, min=1, max=1000)
mode = s.choice('push_mode', ['sync', 'async'], 'sync')
api_key = s.secret('api_key')

HTTP

HttpClient(base_url, auth=None, context=None, timeout=60, ...) makes every REST call:

retries connection errors, 429 (waiting as long as Retry-After says) and 502, 503, 504, for calls that are safe to repeat (GET, PUT, DELETE, and a POST with an idempotency key);
logs in again once on a 401;
sends JSON (amounts as Decimal, dates as ISO 8601) and parses the answer;
refuses plain http:// except for localhost;
raises TargetError with the status and the target's message.

erp = HttpClient(s.str('base_url'), auth=ApiKeyAuth(s.secret('api_key')), context=context, name='ERP')
for vendor in erp.paginate('vendors', params={'modifiedSince': since}, items='data', style='page'):
    ...
erp.post('purchaseInvoices', json=payload, headers={'Idempotency-Key': invoice.invoice_token})

paginate() styles: page (?page=1,2,3), offset, cursor (a token from the previous answer), next_url (a link in the answer, such as @odata.nextLink) and link_header (Link: <...>; rel="next").

Logging in

Helper	For
`ApiKeyAuth(key, header='X-API-Key', prefix='', query_param=None)`	an API key in a header (`prefix='Bearer'` adds the space) or a query parameter
`BasicAuth(username, password)`	HTTP Basic
`BearerAuth(token)`	a fixed bearer token
`OAuth2ClientCredentials(token_url, client_id, client_secret, scope=..., cache=TokenCache(context))`	OAuth 2 client credentials (Microsoft Entra ID, most SaaS)
`OAuth2RefreshToken(context, token_url, client_id, client_secret, refresh_token_secret='refresh_token')`	OAuth 2 refresh token; a new refresh token from the server is stored at once with `context.update_secret`
`OAuth2JwtBearer(token_url, private_key, issuer=..., subject=..., audience=...)`	OAuth 2 with a JWT signed by a private key (assertion grant or private-key client authentication)
`OAuth1Auth(consumer_key, consumer_secret, token, token_secret, realm=...)`	OAuth 1.0a request signing, HMAC-SHA256 (NetSuite token-based authentication) or HMAC-SHA1
`SessionAuth(login_path, login_json=..., token_from='cookie')`	a login call that returns a session cookie or token; logs in again on 401 and logs out at the end (SAP Business One Service Layer)
`client_certificate(context)`	mutual TLS: writes the certificate and key secrets to files only this run can read and returns `(cert_path, key_path)` for `HttpClient(cert=...)`
`soap_client(wsdl, username, password, use_digest=False)`	a zeep SOAP client with WS-Security UsernameToken

Supporting pieces:

TokenCache(context) keeps an access token between runs in the secret access_token_cache, so a run does not fetch a new token each time (some token endpoints allow only a few requests per hour). The secret must be listed under Secrets the connector may update.
expiry_warning(context, key) warns ahead of time when a secret with an expiry date (client_secret_expires_on) or a certificate is about to expire. The warning is in the run log, and in the run result when you pass summary= (the run's RunSummary).
normalize_pem() and Settings.pem() repair a key or certificate pasted into the one-line secret field (the line breaks are rebuilt).

Files

One interface over four stores: list, read_bytes, download, write_bytes, move, delete, exists, makedirs.

Store	For
`SftpStore.from_settings(s)`	an SFTP server outside Nuntiq: settings `sftp_host`, `sftp_port`, `sftp_username`, `sftp_host_key`; secrets `sftp_password`, `sftp_private_key`, `sftp_private_key_passphrase`, `sftp_certificate`. Password, key, key plus password, or an OpenSSH certificate.
`S3Store.from_settings(s)`	the customer's S3 bucket with an access key the customer created (`s3_bucket`, `s3_region`, `s3_prefix`; secrets `s3_access_key_id`, `s3_secret_access_key`)
`context.files()`	the home folder of the run's SFTP user on Nuntiq's own SFTP server, read and write (a `NuntiqFileStore`); no credentials in the connector. See Nuntiq SFTP files below.
`LocalStore(folder)`	a folder on this PC: test file connectors in the toolkit without a server

The SFTP host key is always checked. Without sftp_host_key, or when the server presents another key, the run fails and prints the fingerprint the server offered. Confirm it with the server's owner, then paste it.

Built on the stores:

Inbox(store, incoming, archive=..., error=..., pattern='*.xml') takes files one at a time: it skips files younger than min_age_seconds (still being uploaded), claims each file, hands it to your function, moves it to the archive on success, or to the error folder with <name>.error.txt on failure. A RetryableError puts the file back and stops the run.
ReadOnlyInbox(context, store, incoming, stream) does the same for a source you may not move or delete: it remembers what it took in a delta cursor.
Outbox(store, folder).put(name, data) writes under a temporary name and renames it, so the other side never picks up half a file.
Pgp(context) encrypts, decrypts, signs and verifies with keys from the secrets.
read_csv, read_xlsx, read_xml, write_csv: file contents to rows and back. CSV encoding, BOM and delimiter are detected; XML is parsed safely (no external entities).

store = SftpStore.from_settings(s)
inbox = Inbox(store, 'export', archive='export/archive', error='export/error', pattern='suppliers*.csv')
inbox.process(lambda f: load_suppliers(read_csv(f.read_bytes())))

Nuntiq SFTP files

Nuntiq runs its own SFTP server. A customer's ERP, or a supplier, signs in there as a Data Integration SFTP user and drops or picks up files. A connector run can be given one such user: context.files() then returns that user's home folder, read and write. The connector holds no SFTP password, key or host key for it, and it reaches nothing outside that folder.

files = context.files()                              # a NuntiqFileStore
for f in files.list('in', '*.csv'):
    rows = read_csv(files.read_bytes(f.path))

Choosing the SFTP user. Where the run starts decides where the user is chosen:

The run is started by	The SFTP user is chosen on
an integration rule (`push_invoice` from the process engine)	the rule
a scheduler task of type `connector_dispatch`	the task, under Connector Settings

It is one optional setting: a Data Integration SFTP user of the same customer. Without one, or when the chosen user is disabled, deleted or no longer a Data Integration user, the run still starts, its log carries a warning, and context.files() raises ConfigError with the reason. A configuration export carries the SFTP user's username; an import into another environment finds the user with that username there, also when the user is created after the import.

The same code serves several ERP instances: give each scheduler task or integration rule its own SFTP user.

NuntiqFileStore has the methods of SftpStore and S3Store, plus a few of its own:

Method	Does
`list(folder='', pattern='*', case_sensitive=False)`	the files in one folder (not its subfolders) as `RemoteFile` (`name`, `path`, `size`, `modified`); files still being uploaded under a temporary name (`.filepart`, `.part`, `.tmp`, `.partial`) are left out
`list_folders(folder='')`	the subfolders of a folder
`stat(path)`	`{exists, size, modified, scan_status}` of one file
`exists(path)`	whether the file is there
`read_bytes(path)`	the whole file in memory, at most 50 MB
`download(path, local_path)`	streams the file to disk (for example into `context.work_dir`), any size; returns `local_path`
`write_bytes(path, data, atomic=True, overwrite=True)`	writes a file (`str` as UTF-8), at most 50 MB. A file appears only when it is complete, so every write is atomic. `overwrite=False` refuses a name that exists.
`move(src, dst, overwrite=False)`, `copy(src, dst, overwrite=False)`	within the home folder
`delete(path)`	one file
`makedirs(path)`, `rmdir(path)`	create a folder; remove an empty folder
`backup(path)`	moves the file to the SFTP user's backup area, kept for the user's backup period (7, 30 or 90 days), where the customer can restore it from the portal; returns `{path, backup, retention_days}`

Paths are relative to the SFTP user's home folder, with forward slashes (in/orders.csv; a leading / is ignored). . and .. are refused.

Files are readable only after the malware scan. Every file that arrives on Nuntiq's SFTP server is scanned for malware, also the files a connector writes. read_bytes, download, move, copy and backup work only on a file the scan cleared; while the scan is still running they raise RetryableError and the file stays where it is. The scan usually takes seconds: keep Inbox's min_age_seconds at its default of 60 or more and fresh files are left for the next run anyway. stat(path)['scan_status'] is None while the scan runs and NO_THREATS_FOUND once it cleared the file. A file the scan does not clear is taken out of the folder and the SFTP user's contact is told by email, so the connector does not see it.

File types. Each SFTP user has a list of allowed file extensions. write_bytes, move and copy refuse a name outside that list, and read_bytes and download refuse a file whose name is outside it (for example one renamed after the upload). Inbox writes its <name>.error.txt note only when .txt is allowed for the user; the failed file still moves to the error folder.

Size. Uploads to a Data Integration SFTP user are at most 50 MB per file, so read_bytes holds every file a partner can upload. Use download when you would rather not hold the file in memory (the run has 512 MB).

Error	When
`ConfigError`	the run has no SFTP user, or its SFTP user is disabled, deleted or not a Data Integration user (the message says which)
`RetryableError`	the scan has not cleared the file yet; the file storage did not answer, timed out, answered 408, 429 or 5xx, or the transfer broke off. Nothing changed: try again in a later run.
`FileNotFoundError`	no such file
`PermanentError`	`FILE_TOO_LARGE` (`read_bytes` or `write_bytes` over 50 MB), `FILE_EXISTS` (the target exists and `overwrite` is false), `FILE_TYPE_NOT_ALLOWED`, `SCAN_NOT_CLEAN`, `FOLDER_NOT_EMPTY` (`rmdir`), `INVALID_PATH`

Inbox and Outbox work on it unchanged. An SFTP user whose in folder the ERP fills, with an acknowledgement per file in out:

from lib.helpers import Inbox, Outbox, RunSummary, read_csv, write_csv


def run(context):
    summary = RunSummary(context)
    files = context.files()
    acks = Outbox(files, 'out')

    def handle(f):
        rows = read_csv(f.read_bytes())
        results = [load_order(row) for row in rows]          # your code
        acks.put(f"{f.name}.ack.csv", write_csv(results))

    counts = Inbox(files, 'in', archive='done', error='error', pattern='*.csv',
                   context=context, summary=summary).process(handle)
    return summary.result(**counts)

In the toolkit the folder fixtures/sftp/ is the SFTP user's home folder; see Toolkit overview.

Mapping

Helper	Does
`pick(record, 'a.b[0].c', default)`	a nested value, `default` when any step is missing
`num(value, decimals=None)`	a `Decimal`; reads `1.234,56` and `1,234.56`
`date(value, formats=None)`, `utc_iso(value)`	dates to ISO, timestamps to UTC ISO (for cursors)
`country_code(value)`	ISO 3166 alpha-2 from a code, alpha-3 or an English name
`truthy(value)`, `text(value, max_len)`	yes/no values, trimmed text

Flows

`incremental()`: read only what changed

with incremental(context, 'suppliers') as run:
    for rec in erp.paginate('vendors', params={'modifiedSince': run.since}, items='data'):
        map_supplier(load, rec)
        run.seen(rec['lastModified'])
    load.save_all()

run.since comes from the job parameter since_override, else the cursor the last successful run stored, else now minus the lookback_days setting. The next run starts at the highest value passed to run.seen(). A failed run does not move the cursor. max_records stops a big backlog early (run.full); the next run continues from there. See Delta state.

`run_push()`: the whole `push_invoice`

def push_one(invoice):
    doc = erp.find_invoice(invoice.invoice_token) or erp.create_invoice(to_erp(invoice))
    return PushResult.posted(external_id_1=doc['id'], external_message_1='POSTED')

def run(context):
    return run_push(context, push_one)

run_push serves both callers (the process engine with one invoice, and a scheduler task that claims invoices in batches), handles leases and the dry run, and turns what push_one returns into the right call:

`push_one` returns or raises	Result
`PushResult.posted(...)`	the target booked it: integration result success, with up to three external ids and messages
`PushResult.delivered(...)`	the target has it and decides later: the invoice is acknowledged with the external ids and waits; `get_invoice_status` reports the outcome
`PushResult.failed(code, message)`	the target refused it: integration result failure
`RetryableError`	the target is down: a claimed invoice is given back and offered again next run
any other exception	integration result failure with the error's code

See Async push and status.

`ResultReporter`: outcomes of delivered invoices

reporter = ResultReporter(context)
for inv in reporter.pending():                       # this connector's waiting invoices
    state = erp.get(f"documents/{inv['external_id_1']}")['state']
    if state == 'BOOKED':
        reporter.posted(inv, external_id_2=..., external_message_2='BOOKED')
    elif state == 'REJECTED':
        reporter.failed(inv, 'ERP_REJECTED', 'Rejected in the ERP')
return reporter.result()

An answer can also name the invoice by the id stored at delivery: reporter.posted({'external_id_1': transaction_id}, ...). An invoice without an answer is left alone.

`InvoiceIntake`: invoices into Nuntiq

intake = InvoiceIntake(context)                      # target from the 'target' setting
if intake.target == 'json':
    inv = intake.new_invoice(); ...; intake.submit_invoice(inv)
else:
    intake.submit_file(name, data, attachments=[('delivery_note.pdf', dn_bytes)])

capture sends the PDF through capture, einvoice sends the data file to the customer's inbound file formats, json imports the invoice the connector built. See Invoice ingestion.

Settings and results​

HTTP​

Logging in​

Files​

Nuntiq SFTP files​

Mapping​

Flows​

incremental(): read only what changed​

run_push(): the whole push_invoice​

ResultReporter: outcomes of delivered invoices​

InvoiceIntake: invoices into Nuntiq​