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() <- 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?" answer (93 -> 100 or 97)
If your code crashes between 4 and 5, the invoice sits at status 93
("PendingResult") until either you run again and call
integration_result (idempotent on success), or the lock expires and
the invoice falls back to status 90/91 for re-claim by another run.
claim()
leases = load.claim(
state='PendingIntegration', # only valid value today
limit=10, # 1..50
lease_ttl_seconds=300, # 60..1800
idempotency_key=None, # see below
ext_reference_1=None, # filter — exact match (case-insensitive)
# ext_reference_2..5=None,
)
Returns a list of InvoiceLease objects:
lease.invoice_token— pass this to all subsequent stepslease.lease_expires_at— when the lock expires (someone else can re-claim)lease.lease_token— id of this specific lease
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.
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), group, label |
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.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')
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.invoice_token)
Idempotent. Moves the invoice from PendingAcknowledgement (92) to
PendingResult (93). Call it once you've successfully started the ERP
push but before you have a final outcome.
integration_result()
# success path
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 and external_message_2/3 also available
)
# failure path
load.integration_result(
lease.invoice_token,
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.
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).
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'])
list_all() doesn't change any state.
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: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 to use on 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, combine this with delta_run.