The invoice claim flow

InvoiceLoad is different from every other Load: invoices in Nuntiq are produced by the document-processing pipeline (email intake, manual upload, API import), not created by connectors. Connectors consume them via a claim-based workflow.

from lib.objects.invoice import InvoiceLoad
load = InvoiceLoad(context)

The five steps

   1. claim()              ← locks N invoices for you  (90/91 → 92)
   2. get_by_token()       ← read full detail for each
   3. push to ERP          ← your code
   4. acknowledge()        ← marks "received by integrator"  (92 → 93)
   5. integration_result() ← final "did it post?"  (93 → 100 success | 97 failed)

If your code crashes between (4) and (5), the invoice sits at status 93 ("PendingResult") until either you run again and call integration_result (which is idempotent on success), or the lease expires and the invoice falls back to 90/91 for re-claim by another run.

1. claim()

leases = load.claim(
    state='PendingIntegration',     # only valid value today
    limit=10,                       # 1..50
    lease_ttl_seconds=300,          # 60..1800
    idempotency_key=None,
    ext_reference_1=None,           # optional filter (exact, case-insensitive)
    # ext_reference_2..5=None,
)

Returns a list of InvoiceLease objects:

• lease.invoice_token — pass this to every subsequent step
• lease.lease_expires_at — when the lock expires
• lease.lease_token — id of this specific lease

claim() is atomic. The same invoice cannot be returned to two workers running this entrypoint concurrently — that's the whole point of the lease model. If your code crashes before acknowledge(), the lease expires (default 5 min) and the next run picks the invoice back up.

Idempotency: pass the same idempotency_key on retry within the TTL and you get the same invoices back. Safe to retry on a network blip without locking different invoices.

2. get_by_token()

invoice = load.get_by_token(lease.invoice_token)

Returns a read-only Invoice with the full denormalized graph. Header fields, lines, addresses, taxes, charges, discounts, attachments, AI matches, and the origin email — all in one call.

Attribute	Type	Notes
invoice.invoice_token	str (UUID)	The public handle
invoice.status	str	PendingIntegration, PendingResult, ...
invoice.status_info	{code, group, label}	Numeric + symbolic status
invoice.fields	list[InvoiceField]	Header fields, template-driven
invoice.lines	list[InvoiceLine]	Each line has its own .fields
invoice.addresses	list[InvoiceAddress]	Up to 4 — vendor, remit_to, ship_to, bill_to
invoice.taxes	list[InvoiceTax]	Tax breakdown rows
invoice.charges	list[InvoiceCharge]	Fees, surcharges
invoice.discounts	list[InvoiceDiscount]	Discount terms
invoice.attachments	list[InvoiceAttachment]	PDFs, source email, etc.
invoice.ai_matches	list[InvoiceAIMatch]	AI-suggested PO matches
invoice.email	`InvoiceEmail	None`

Reading fields

Header and line fields are template-driven — the keys come from the customer's invoice template configuration and are typically camelCase. The defaults shipped by Nuntiq use keys like invoiceNumber, invoiceDate, grossAmount, taxAmount, netAmount, currencyCode, poNumber. Customer templates can override or extend these, so the keys your code calls get_field() with must match what the customer's template emits.

inv_number = invoice.get_field('invoiceNumber')
gross      = invoice.get_field('grossAmount')
currency   = invoice.get_field('currencyCode')
po_number  = invoice.get_field('poNumber')

for line in invoice.lines:
    qty   = line.get_field('quantity')
    price = line.get_field('unitPrice')
    amt   = line.get_field('netAmount')

vendor = invoice.get_address('vendor')
if vendor:
    print(vendor.address_name, vendor.street, vendor.city)

Don't hardcode field-key strings if you can avoid it — iterate invoice.fields if you need everything, or check the customer's template configuration before writing the connector to see which keys are actually mapped.

3. Push to your target

Whatever that looks like. REST POST, SOAP envelope, SFTP file, EDI message. See Integration patterns for end-to-end worked examples.

4. acknowledge()

load.acknowledge(lease.invoice_token)

Idempotent. Moves the invoice from status 92 (PendingAcknowledgement) to 93 (PendingResult). Call this before integration_result — the API rejects a result on an invoice that isn't in 93.

A common pattern is "acknowledge as soon as the external system confirmed receipt, even before you have a final outcome". That way if your code crashes between receiving the external system's response and reporting it to Nuntiq, the invoice doesn't fall back to PendingIntegration and get pushed twice.

5. integration_result()

Success:

load.integration_result(
    lease.invoice_token,
    success=True,
    external_id_1='ERP-INV-12345',     # the ERP-side invoice id
    external_message_1='POSTED',       # short status string
    # external_id_2/3 + external_message_2/3 are also available
)

Advances the invoice to 100 (Processed).

Failure:

load.integration_result(
    lease.invoice_token,
    success=False,
    failure_code='LINE_MISMATCH',      # business-level code, your taxonomy
    failure_message='Line 3 did not match any PO line',
)

Advances to 97 (IntegrationFailed). failure_code and failure_message appear in the Nuntiq admin UI for whoever investigates.

Overriding a previous success: if the invoice is already Processed and you call with success=False, you get a 409 unless you also pass force_override=True. Rare — usually means you're correcting a bad earlier outcome.

Listing without claiming

Sometimes you just want to see what's pending without locking it:

for record in load.list_all(state='PendingIntegration'):
    print(record['invoice_token'])

list_all() doesn't change any state.

Cross-invoice lifecycle polling

Separate from the claim flow. Use this when you want to react to changes on invoices over time, not push them outward.

for msg in load.lifecycle_messages_since(
    since='2026-04-25T00:00:00',         # ISO-8601 (naked or +offset)
    tz='Europe/Amsterdam',               # required, IANA zone
    ext_references={1: 'CUSTOMER-A'},    # at least one required
):
    context.logger.info(f"{msg['recorded_at']}  {msg['code']}  {msg['invoice_token']}")

The watermark for the next poll is msg['recorded_at'] of the last message returned. Filtering is case-insensitive equality on ext_reference_1..5 — no wildcards.

For a full incremental cursor with stuck-run handling, wrap this in delta_run.

What's next

• SFTP push pattern — full worked example using the claim flow.
• BaseLoad — the read/write patterns that apply to *non-*invoice entities.
• Invoice reference — field-by-field.