"""
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())
