InvoiceLoad

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.

Import

from lib.objects.invoice import InvoiceLoad

The claim flow

Five steps, in order:

   1. claim()                    <- leases N invoices to you (90/91 -> 92)
   2. get_by_token()             <- read full detail for each
   3. push to ERP                <- your code
   4. acknowledge(lease)         <- "the ERP has it" (92 -> 93)
   5. integration_result(lease)  <- final "did it post?" answer (-> 100 or 97)

release(lease) gives a claimed invoice back instead. The concept page Invoice claim flow explains leases, re-offers and the error codes.

Methods

Method	What it does
claim(state='PendingIntegration', limit=10, lease_ttl_seconds=300, idempotency_key=None, ext_reference_1..5=None)	Lease up to limit (1..50) invoices for lease_ttl_seconds (60..1800). Returns a list of InvoiceLease
get_by_token(invoice_token)	The full Invoice
acknowledge(invoice, lease_token=None)	92 to 93. Returns {status: 'acknowledged' or 'unchanged', invoice_token, previous_status, new_status}
release(invoice, lease_token=None, reason=None)	92 or 93 back to 91, offered again. Returns {status: 'released', ...}
integration_result(invoice, success=True, external_id_1..3=None, external_message_1..3=None, failure_code=None, failure_message=None, force_override=False, debug=False, lease_token=None)	Success: 100. Failure: 97. Returns {status: 'updated' or 'unchanged', ...}
acknowledge_many(leases)	Acknowledge many; a result per invoice
release_many(leases, reason=None)	Release many; a result per invoice
integration_results(results)	Report many results; a result per invoice
list(state='PendingIntegration', page_size=100, cursor=None, updated_since=None, **filters)	One page of invoice tokens: {'data': [...], 'next_cursor': ...}
list_all(state='PendingIntegration', page_size=100, updated_since=None, **filters)	Every page (generator)
find_by_external_id(external_id_1=None, external_id_2=None, external_id_3=None, state=None, page_size=100)	Invoices by the external ids a result stored (generator)
lifecycle_messages_since(since, ext_references, ...)	The cross-invoice lifecycle feed (generator)

invoice is an InvoiceLease (its lease_token is used), or an invoice token string together with lease_token=. While an invoice is claimed (92) or acknowledged (93), these calls need the lease token: without it the API answers 409 lease_token_required, with a lease that another claim has replaced 409 lease_mismatch (do not post the invoice then). Invoices that were never claimed take a result without a lease.

The bulk methods take up to 100 invoices per request (longer lists are split), handle each invoice on its own and return one result per invoice (index, invoice_token, status, previous_status, new_status, errors). They do not raise for a rejected invoice; check the results.

claim()

leases = load.claim(
    state='PendingIntegration',         # the only state a claim takes
    limit=10,                           # 1..50
    lease_ttl_seconds=300,              # 60..1800
    idempotency_key=None,               # see below
    ext_reference_1=None,               # filter: case-insensitive pattern, % wildcards
    # ext_reference_2..5=None,
)

Returns a list of InvoiceLease objects:

Attribute	Notes
lease.invoice_token	The invoice
lease.lease_token	Sent with every later call for the invoice when you pass the lease
lease.lease_expires_at	When the lease runs out. After that, another claim can take the invoice
lease.reoffered	True when an earlier claim's lease had run out: check the target system before posting

A claim takes invoices at 90/91 and claimed invoices whose lease ran out.

Idempotency: with an idempotency_key, a retry returns your live leases for that key first (with their time extended), then takes new invoices up to limit. Safe to retry after a network error. Without a key a claim takes only free invoices.

get_by_token()

invoice = load.get_by_token(lease.invoice_token)

Returns an Invoice object (read-only). Has a fixed shape:

Attribute	Type
invoice.id	int, internal id
invoice.invoice_token	UUID string
invoice.email_token	UUID string (the email it was extracted from, if any)
invoice.status	string group: PendingIntegration, PendingResult, Processed, ...
invoice.status_info	dict with code (numeric), key, group, label, description
invoice.fields	list of {key, label, value}: header fields, template-driven
invoice.lines	list of InvoiceLine, each with its own .fields
invoice.addresses	list of InvoiceAddress: vendor / remit_to / ship_to / bill_to
invoice.taxes	list of InvoiceTax
invoice.charges	list of InvoiceCharge
invoice.discounts	list of InvoiceDiscount
invoice.deductions	list of InvoiceDeduction: retainage, withholding tax, deposits
invoice.amount_due	the amount payable after deductions (header field amountDue), or None
invoice.attachments	list of InvoiceAttachment (the PDF, original email, etc.)
invoice.ai_matches	list of InvoiceAIMatch: AI-suggested PO matches
invoice.email	InvoiceEmail or None

Reading fields

Header and line fields are template-driven, so the keys come from the customer's invoice template config and are typically camelCase. Nuntiq's default templates use invoiceNumber, invoiceDate, grossAmount, netAmount, taxAmount, currencyCode, poNumber, unitPrice, etc. Customer-specific templates can override or add to these.

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

Deductions and amount due

Some invoices subtract amounts after the total: retainage held back on a progress bill, withholding tax (the Italian ritenuta d'acconto, the Spanish IRPF, Mexican ISR or IVA retenido) or a deposit already paid. grossAmount is the total before them; invoice.amount_due is what is left to pay (gross minus the deductions). A deduction is not a discount and not negative tax: post it to its own account or field in the target system.

for d in invoice.deductions:
    print(d.deduction_type,          # retainage | withholding | deposit | other
          d.deduction_description,   # as printed, e.g. 'Retainage 10%'
          d.deduction_rate,          # percentage, when printed
          d.deduction_base_amount,   # amount the rate applies to
          d.deduction_amount)        # amount deducted (negative on a credit note)

to_pay = invoice.amount_due if invoice.amount_due is not None else invoice.get_field('grossAmount')

amount_due is None when the invoice prints no separate amount due (then the gross amount is payable) or when the customer's template has no amountDue field.

For lines:

for line in invoice.lines:
    print(line.get_field('description'),
          line.get_field('quantity'),
          line.get_field('unitPrice'),
          line.get_field('netAmount'))

For addresses:

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

acknowledge()

load.acknowledge(lease)

Moves the invoice from PendingAcknowledgement (92) to PendingResult (93); a second call answers unchanged. Call it once the ERP confirmed receipt, before you have a final outcome. An acknowledged invoice is never offered to another claim. Keep the lease: the result needs it.

integration_result()

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

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

success=True advances to Processed (100). success=False advances to IntegrationFailed (97) and the failure_code + failure_message appear in the Nuntiq admin UI. Either way the lease is cleared.

A failed invoice can be rescued later with success=True. If the invoice is already Processed and you call with success=False, you get 409 already_processed unless you also pass force_override=True (rare: usually you're correcting a bad earlier outcome). debug=True adds the numeric status codes to the answer.

release()

load.release(lease, reason='ERP closed for month-end')

Gives a claimed (92) or acknowledged (93) invoice back: status 91, lease cleared, offered to the next claim.

Bulk

load.acknowledge_many(leases)
load.release_many(leases, reason='shutting down')
results = load.integration_results([
    {'lease': lease, 'success': True, 'external_id_1': erp_id},
    {'invoice_token': token, 'lease_token': lease_token, 'success': False,
     'failure_code': 'LINE_MISMATCH', 'failure_message': 'Line 3 did not match'},
])

Each item is an InvoiceLease, or a dict with lease (an InvoiceLease) or invoice_token + lease_token, plus the fields of the single call.

Nested objects on an invoice

invoice.attachments, invoice.ai_matches, and invoice.email are returned automatically by get_by_token(). They're typed objects, not raw dicts. Here's what's on each.

InvoiceAttachment: files attached to the invoice

Returned for every PDF, image, and email that's tied to the invoice. Use the helpers on the object to actually pull the file bytes; there's no need to construct your own URL.

Field	Type	Notes
attachment_token	str (UUID)	Pass to download_attachment if you call the low-level API directly
file_name	str	Original filename (e.g. invoice-1234.pdf)
content_type	str	MIME type (application/pdf, image/png, message/rfc822, …)
type	str	One of INVOICE_IMAGE (the primary PDF), EMAIL_ATTACHMENT (a sibling file in the same email), EMAIL (the source .eml itself)
parent_token	str (UUID) or None	If this attachment was derived from another (e.g. PDF rendered from a .docx), the token of the source

Three methods to read the bytes; pick whichever fits your downstream call:

for att in invoice.attachments:
    # Option A: save to a per-run scratch path under context.work_dir
    local_path = att.save()                  # uses att.file_name
    local_path = att.save('renamed.pdf')     # override the filename
    # `local_path` is the absolute path of the saved file

    # Option B: in-memory bytes
    raw = att.read_bytes()                   # returns `bytes`

    # Option C: base64 string (handy for JSON payloads to third-party APIs)
    b64 = att.read_base64()                  # returns a str

save() writes under context.work_dir, which is wiped at end of run in production. See Work directory for what survives and what doesn't. Path-traversal in the override filename is rejected: you can't escape work_dir.

The attachment objects are pre-wired to the JobContext that returned them, so the three methods work without you having to pass anything.

InvoiceAIMatch: AI-suggested PO / receipt matches

When the platform's enrichment pipeline thinks an invoice line matches a specific PO line or receipt line, it records the suggestion as an InvoiceAIMatch row. Use these as inputs to your integration logic if you trust the model; otherwise treat them as hints and validate before posting.

Field	Type	Notes
id	int	Internal id
invoice_id	int	Parent invoice id (matches invoice.id)
purchase_order_number	str	The PO this line is matched against
purchase_order_line_number	int	The matched PO line
product_code	str	Suggested product code from the matched PO line
receipt_number	str	Matched receipt, if any
receipt_line_number	int	Matched receipt line, if any
quantity	float	Quantity the AI extracted from the invoice
unit_price	float
unit_of_measure	str
amount	float	Line net amount
confidence_score	float	0.0–1.0. Higher = more confident.
reason	str	Free-form text the model used to justify the match

for m in invoice.ai_matches:
    if m.confidence_score is None or m.confidence_score < 0.85:
        continue   # too uncertain to auto-act on
    push_to_erp(
        po_number=m.purchase_order_number,
        po_line=m.purchase_order_line_number,
        receipt_line=m.receipt_line_number,
        qty=m.quantity,
    )

Multiple suggestions per line are possible; iterate and pick whichever suits your business rule.

InvoiceEmail: the source email metadata

Present only when the invoice arrived via email intake (the most common case). None when the invoice was uploaded manually or via API. Use it when you need to surface the original sender or reply on top of the same thread.

Field	Type	Notes
email_token	str (UUID)	The email's public token
eml_from	str	Raw From: header value
eml_to	str	Raw To: header; may include multiple addresses, comma-separated
eml_replyto	str or None	Reply-To: if the sender set one
eml_subject	str or None
eml_body	str or None	The text body, plain-text if available
eml_received	str or None	ISO-8601 timestamp when intake received the message

if invoice.email and invoice.email.eml_from:
    notify_finance(
        f"Invoice {invoice.invoice_token} from {invoice.email.eml_from} "
        f"received at {invoice.email.eml_received}"
    )

eml_to and eml_from are unparsed. If you need the bare address, run them through email.utils.parseaddr() from the Python stdlib.

Listing without claiming

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

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

# What changed since the last run
for record in load.list_all(state='PendingResult', updated_since='2026-10-01T00:00:00Z'):
    ...

# Find the invoice an ERP document belongs to
for record in load.find_by_external_id(external_id_1='ERP-INV-12345'):
    print(record['invoice_token'], record['status_group'])

list() and list_all() don't change any state. Each record has invoice_token, status, status_group, updated_at and external_id_1..3. Paging follows next_cursor, ordered by (updated_at, id).

state	Invoices
PendingIntegration	What a claim can take: 90/91, and 92 whose lease ran out
PendingAcknowledgement	92
PendingResult	93
PendingConnectorIntegration	94: handed to a connector's push_invoice job by the process engine. For housekeeping, never for claiming

Other filters: ext_reference_1..5 (case-insensitive pattern, % wildcards) and external_id_1..3 (exact).

Cross-invoice lifecycle delta 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:00Z',         # ISO-8601, UTC
    ext_references={1: 'CUSTOMER-A'},     # at least one required
    tz='UTC',                             # only read when `since` has no offset
):
    context.logger.info(f"{msg['recorded_at']}  {msg['code']}  {msg['invoice_token']}")

since is inclusive. For the next poll, start from the recorded_at of the last message you processed and skip the message ids you already handled. Filtering is case-insensitive equality on ext_reference_1..5, with no wildcards. See Lifecycle messages.

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