> ## Documentation Index
> Fetch the complete documentation index at: https://docs.watchdog.no/llms.txt
> Use this file to discover all available pages before exploring further.

# Upload invoices

> Send an invoice file to Watchdog and get the result of its checks.

Uploading an invoice takes two steps: upload the file and start the import. Watchdog does the rest
on its own. If you want to follow the invoice until it has been checked, add the two optional steps.
The [Python example](#python-example) at the bottom does all four and can be used as it is.

## Before you start

You need an API key with **Write** access, from **Settings → Personal → API keys** in Watchdog, and
your organization ID, which is the `org_…` part of the address when your organization is open in
Watchdog. Send both with every request:

```bash theme={null}
export API_URL=https://api-canary.watchdog.no
export API_KEY='<your API key>'
export ORGANIZATION_ID='<your org_… ID>'
```

<Note>
  Use `api-canary.watchdog.no` until `api.watchdog.no` moves over to the new API. After that, you
  only need to change the address.
</Note>

## 1. Upload the file

Register the file, then send it to the upload address you get back, with exactly the headers you
get back and without your API key.

```bash theme={null}
curl -s "$API_URL/v1/documents/upload" \
  -H "Authorization: Bearer $API_KEY" -H "X-Organization-Id: $ORGANIZATION_ID" \
  -H 'Content-Type: application/json' \
  -d "{\"file_name\": \"invoice.pdf\", \"mime_type\": \"application/pdf\", \"file_size\": $(wc -c < invoice.pdf)}" \
  > upload.json

jq -r '.headers | to_entries[] | "\(.key): \(.value)"' upload.json > upload.headers
curl -s --upload-file invoice.pdf --header @upload.headers "$(jq -r .upload_url upload.json)"
```

## 2. Start the import

```bash theme={null}
curl -s "$API_URL/v1/invoices/imports" \
  -H "Authorization: Bearer $API_KEY" -H "X-Organization-Id: $ORGANIZATION_ID" \
  -H 'Content-Type: application/json' \
  -d "{\"primary_document_id\": \"$(jq -r .document_id upload.json)\"}" \
  > import.json
```

Attachments are uploaded the same way and listed in `attachment_document_ids`.

That's it: the invoice is on its way into Watchdog.

## 3. Wait for the import (optional)

Only needed if you want the invoice ID or the result of the check.

Read the import every few seconds until it's no longer `queued` or `running`:

```bash theme={null}
curl -s "$API_URL/v1/invoices/imports/$(jq -r .id import.json)" \
  -H "Authorization: Bearer $API_KEY" -H "X-Organization-Id: $ORGANIZATION_ID"
```

When it's `completed` with the outcome `imported`, `invoice_id` is your invoice. With `duplicate`, it
points to the invoice Watchdog already had. With `not_invoice`, the file wasn't an invoice and there's
nothing to follow.

## 4. Wait for the check (optional)

Read the invoice every few seconds until `workflows.settled` is `true`. Then `alerts` has the
result of checking it against your agreements.

```bash theme={null}
curl -s "$API_URL/v1/invoices/$INVOICE_ID" \
  -H "Authorization: Bearer $API_KEY" -H "X-Organization-Id: $ORGANIZATION_ID"
```

## Good to know

* The invoice must be a PDF or XML file (Peppol BIS Billing 3.0 or EHF 2.0). Attachments can be any
  file type.
* Sending the same invoice twice is safe: you get the existing invoice back.
* Send an `Idempotency-Key` header when starting an import. If the request times out, send it again
  with the same key and you get the same import back instead of a new one.
* Include `error.request_id` when you contact us about an error.

## Python example

The script uploads an invoice with any attachments, waits for the import and the check, and prints
the result. It retries temporary errors on its own. It needs Python 3.9 or newer and `requests`.

```bash theme={null}
pip install requests
export WATCHDOG_API_KEY='<your API key>'
export WATCHDOG_ORGANIZATION_ID='<your org_… ID>'
python watchdog_upload_example.py invoice.pdf price-list.xlsx
```

```python watchdog_upload_example.py expandable theme={null}
"""
Upload an invoice to Watchdog and wait until Watchdog has checked it.

The flow:

  1. Upload each file: register it, then send     POST /v1/documents/upload
     its bytes to the returned upload URL         PUT  <upload_url>
  2. Start an invoice import                      POST /v1/invoices/imports
  3. Wait until the import has finished           GET  /v1/invoices/imports/{id}
  4. Wait until the invoice has been checked      GET  /v1/invoices/{id}

Requirements: Python 3.9+ and `pip install requests`.

Usage:
  export WATCHDOG_API_KEY="..."
  export WATCHDOG_ORGANIZATION_ID="..."   # From the Watchdog URL: app.watchdog.no/<organization id>/...
  python watchdog_upload_example.py invoice.pdf [attachment.xlsx ...]

Optional setting:
  WATCHDOG_API_URL  API address (default: https://api-canary.watchdog.no)
"""

import mimetypes
import os
import sys
import time
import uuid
from pathlib import Path
from typing import Optional

import requests

API_URL = os.environ.get("WATCHDOG_API_URL", "https://api-canary.watchdog.no").rstrip("/")
API_KEY = os.environ.get("WATCHDOG_API_KEY", "")
ORGANIZATION_ID = os.environ.get("WATCHDOG_ORGANIZATION_ID", "")

POLL_INTERVAL_SECONDS = 10
POLL_TIMEOUT_SECONDS = 30 * 60
MAX_ATTEMPTS = 5  # Per request, for temporary errors: rate limits, brief outages, network failures.

# The invoice itself must be a PDF or XML file; otherwise the import fails with an explanation.
# Attachments can be any file type, such as Excel, Word or email. Watchdog reads the type from
# each file's contents, so the type sent below only needs to be a reasonable guess.


class WatchdogError(Exception):
    """An error answer from the Watchdog API. Quote `request_id` when contacting support."""

    def __init__(self, status: int, body: dict):
        error = body.get("error", {})
        self.status = status
        self.code = error.get("code", "unknown")
        self.request_id = error.get("request_id")
        details = error.get("description") or error.get("message") or "No details"
        super().__init__(f"{status} {self.code}: {details} (request_id: {self.request_id})")


session = requests.Session()
session.headers["Authorization"] = f"Bearer {API_KEY}"
session.headers["X-Organization-Id"] = ORGANIZATION_ID  # The organization the invoice belongs to.


def retry_delay(response: Optional[requests.Response], attempt: int) -> float:
    """Wait as long as the API asks (Retry-After), otherwise back off exponentially."""
    retry_after = response.headers.get("Retry-After", "") if response is not None else ""
    return float(retry_after) if retry_after.isdigit() else 2**attempt


def send(request) -> requests.Response:
    """Send a request and return the successful response.

    Network failures, rate limits (429) and temporary unavailability (500, 502, 503, 504) are
    retried a few times. Other errors are raised as WatchdogError straight away.
    """
    attempt = 1
    while True:
        try:
            response = request()
        except (requests.ConnectionError, requests.Timeout):
            if attempt == MAX_ATTEMPTS:
                raise
            response = None
        if response is not None and response.ok:
            return response
        if response is not None and response.status_code not in (429, 500, 502, 503, 504):
            raise WatchdogError(response.status_code, error_body(response))
        if attempt == MAX_ATTEMPTS:
            raise WatchdogError(response.status_code, error_body(response))
        time.sleep(retry_delay(response, attempt))
        attempt += 1


def call_api(method: str, path: str, **kwargs) -> dict:
    """Call the API and return its JSON body."""
    return send(lambda: session.request(method, f"{API_URL}{path}", timeout=60, **kwargs)).json()


def error_body(response: requests.Response) -> dict:
    """The API's error JSON, or an empty body when a proxy answered with something else."""
    try:
        return response.json()
    except ValueError:
        return {}


def upload_file(path: Path) -> str:
    """Step 1: upload one file and return its document id."""
    mime_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    content = path.read_bytes()
    upload = call_api(
        "POST",
        "/v1/documents/upload",
        json={
            "file_name": path.name,
            "mime_type": mime_type,
            "file_size": len(content),
            "category": "invoice",
        },
    )
    # Send exactly the headers Watchdog returned, and not your API key: this goes to storage.
    try:
        send(lambda: requests.put(upload["upload_url"], data=content, headers=upload["headers"], timeout=300))
    except WatchdogError as error:
        if error.status != 412:  # 412: an earlier attempt already stored the file.
            raise
    return upload["document_id"]


def start_import(primary_document_id: str, attachment_document_ids: list) -> dict:
    """Step 2: start the import and return it.

    The Idempotency-Key lets call_api retry safely: the same key never starts a second import.
    """
    return call_api(
        "POST",
        "/v1/invoices/imports",
        headers={"Idempotency-Key": str(uuid.uuid4())},
        json={
            "primary_document_id": primary_document_id,
            "attachment_document_ids": attachment_document_ids,
        },
    )


def wait_for_import(invoice_import: dict) -> dict:
    """Step 3: poll the import until it has finished.

    A file Watchdog already has comes back finished straight away, as a duplicate.
    """
    deadline = time.monotonic() + POLL_TIMEOUT_SECONDS
    while invoice_import["status"] in ("queued", "running"):
        if time.monotonic() > deadline:
            raise TimeoutError(f"Import {invoice_import['id']} is still {invoice_import['status']}")
        print(f"   import is {invoice_import['status']}")
        time.sleep(POLL_INTERVAL_SECONDS)
        invoice_import = call_api("GET", f"/v1/invoices/imports/{invoice_import['id']}")
    return invoice_import


def wait_for_invoice(invoice_id: str) -> dict:
    """Step 4: poll the invoice until matching and the alert check have settled."""
    deadline = time.monotonic() + POLL_TIMEOUT_SECONDS
    while True:
        invoice = call_api("GET", f"/v1/invoices/{invoice_id}")
        if invoice["workflows"]["settled"]:
            return invoice
        if time.monotonic() > deadline:
            raise TimeoutError(f"Invoice {invoice_id} has not settled yet")
        print(f"   alert check is {invoice['workflows']['alert_check']['status']}")
        time.sleep(POLL_INTERVAL_SECONDS)


def print_summary(invoice: dict) -> None:
    alerts = invoice["alerts"]
    counts = alerts["counts"]
    supplier = invoice["supplier"]["name"] or invoice["supplier"]["invoiced_as"]["name"]
    print()
    print(f"Invoice:      {invoice['invoice_number']} from {supplier}")
    print(f"Total:        {invoice['amounts']['including_vat']} {invoice['amounts']['currency_code']}")
    print(f"Alert status: {alerts['status']}")
    print(f"Open alerts:  {counts['pending'] + counts['claimed']}")
    print(f"Agreements:   {alerts['agreements']['checked']} of {alerts['agreements']['total']} checked")
    print(f"In Watchdog:  {invoice['url']}")


def import_invoice(primary: Path, attachments: list) -> Optional[dict]:
    """Run the whole flow. Returns the checked invoice, or None when no invoice was created."""
    print(f"1. Uploading {primary.name}" + (f" and {len(attachments)} attachment(s)" if attachments else ""))
    primary_id = upload_file(primary)
    attachment_ids = [upload_file(path) for path in attachments]

    print("2. Starting the import")
    invoice_import = start_import(primary_id, attachment_ids)

    print(f"3. Waiting for import {invoice_import['id']}")
    invoice_import = wait_for_import(invoice_import)
    if invoice_import["status"] != "completed":
        failure = invoice_import["failure"] or {}
        print(f"   The import {invoice_import['status']}: {failure.get('message', 'no details')}")
        return None
    if invoice_import["outcome"] == "not_invoice":
        print(f"   Not an invoice: {invoice_import['classification_reason']}")
        return None
    if invoice_import["outcome"] == "duplicate":
        print("   Watchdog already has this invoice; following the existing one")

    print(f"4. Waiting for Watchdog to check invoice {invoice_import['invoice_id']}")
    invoice = wait_for_invoice(invoice_import["invoice_id"])
    print_summary(invoice)
    return invoice


def main() -> int:
    if not API_KEY or not ORGANIZATION_ID:
        print("Set WATCHDOG_API_KEY and WATCHDOG_ORGANIZATION_ID first.", file=sys.stderr)
        return 2
    if len(sys.argv) < 2:
        print("Usage: python watchdog_upload_example.py invoice.pdf [attachment.xlsx ...]", file=sys.stderr)
        return 2
    primary, *attachments = (Path(argument) for argument in sys.argv[1:])
    try:
        invoice = import_invoice(primary, attachments)
    except WatchdogError as error:
        if error.code == "invoice_processing_capacity_exhausted":
            print("Your organization has no invoice processing capacity left. Contact Watchdog.", file=sys.stderr)
        else:
            print(f"Watchdog API error: {error}", file=sys.stderr)
        return 1
    return 0 if invoice else 1


if __name__ == "__main__":
    sys.exit(main())
```