Best Practices Overview

Proven patterns and recommendations for building robust Peppol integrations.

Core Principles

1. Validate Before Sending

Always validate invoices before transmission:

# Always validate first
validation = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/documents/validate",
    headers={"X-API-Key": api_key, "Content-Type": "application/json"},
    json={"document": invoice_xml},
)

if validation.json()["valid"]:
    # Then send - 202 Accepted, queued for asynchronous delivery
    response = requests.post(
        "https://app.goroute.ai/peppol-api/api/v1/documents",
        headers={"X-API-Key": api_key, "Content-Type": "application/json"},
        json={
            "receiver_scheme": "9959",
            "receiver_id": "987654321",
            "document": invoice_xml,
        },
    )

Both calls take JSON, and the UBL XML travels as a string in the document field rather than as a raw XML request body. receiver_scheme, receiver_id and document are the required fields on the send call; sender_scheme and sender_id default to your organization.

Why: Prevents rejections, saves time, improves delivery rates.

Earlier revisions of this page documented endpoints that do not exist

This page previously showed a send path and a validate path sitting directly under /api/v1, and a single-call batch path alongside them. None of the three has ever existed, and code copied from those samples has never worked. The real endpoints are POST /api/v1/documents, POST /api/v1/documents/validate, and the two-step batch resource under /api/v1/batches.

2. Handle Errors Gracefully

Implement comprehensive error handling:

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60)
)
def send_invoice(invoice_data):
    try:
        response = requests.post(url, json=invoice_data)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 400:
            # Validation error - don't retry
            logging.error(f"Validation failed: {e.response.json()}")
            raise
        elif e.response.status_code == 429:
            # Rate limited - retry with backoff
            logging.warning("Rate limited, retrying...")
            raise
        else:
            # Other errors
            logging.error(f"API error: {e}")
            raise

3. Use Idempotency

Prevent duplicate sends with idempotency keys:

import uuid

# Generate unique idempotency key
idempotency_key = str(uuid.uuid4())

response = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/documents",
    headers={
        "X-API-Key": api_key,
        "Content-Type": "application/json",
        "Idempotency-Key": idempotency_key
    },
    json={
        "receiver_scheme": "9959",
        "receiver_id": "987654321",
        "document": invoice_xml
    }
)

# Same key = same result if retried. The 202 response echoes the key back as
# `idempotency_key`, alongside `transaction_id`, `status`, `message` and `created_at`.

4. Monitor Delivery Status

Track every document to completion:

def track_until_delivered(transaction_id, timeout_minutes=30):
    """Poll for delivery status."""
    import time
    
    start = time.time()
    while time.time() - start < timeout_minutes * 60:
        status = requests.get(
            f"https://app.goroute.ai/peppol-api/api/v1/transactions/{transaction_id}",
            headers={"X-API-Key": api_key}
        ).json()
        
        if status["status"] == "delivered":
            return {"success": True, "details": status}
        elif status["status"] == "failed":
            return {"success": False, "error": status["error"]}
        
        time.sleep(30)  # Check every 30 seconds
    
    return {"success": False, "error": "timeout"}

Integration Architecture

Recommended Flow

Queue-Based Architecture

For high volumes, use a message queue:

# Producer: Queue invoice for processing
import redis
import json

r = redis.Redis()

def queue_invoice(invoice):
    r.lpush("invoice_queue", json.dumps({
        "id": invoice["id"],
        "xml": invoice["xml"],
        "retry_count": 0,
        "queued_at": datetime.utcnow().isoformat()
    }))

# Consumer: Process queued invoices
def process_queue():
    while True:
        item = r.brpop("invoice_queue", timeout=30)
        if item:
            invoice = json.loads(item[1])
            try:
                send_invoice(invoice)
            except Exception as e:
                handle_failure(invoice, e)

Data Quality

Clean Input Data

def clean_invoice_data(data):
    """Sanitize invoice data before sending."""
    
    # Trim whitespace
    data["seller_name"] = data["seller_name"].strip()
    data["buyer_name"] = data["buyer_name"].strip()
    
    # Normalize identifiers
    data["vat_number"] = data["vat_number"].upper().replace(" ", "")
    
    # Validate amounts
    data["total"] = round(data["total"], 2)
    
    # Ensure required fields
    if not data.get("buyer_reference") and data["is_b2g"]:
        raise ValueError("Buyer reference required for B2G")
    
    return data

Validate Before Queuing

def pre_validate(invoice_data):
    """Quick validation before expensive API call."""
    
    errors = []
    
    # Check required fields
    required = ["invoice_id", "issue_date", "seller", "buyer", "lines"]
    for field in required:
        if not invoice_data.get(field):
            errors.append(f"Missing required field: {field}")
    
    # Check identifier formats
    if not validate_peppol_id(invoice_data["buyer"]["scheme"], 
                               invoice_data["buyer"]["identifier"]):
        errors.append("Invalid buyer Peppol ID")
    
    # Check totals match
    calculated_total = sum(line["amount"] for line in invoice_data["lines"])
    if abs(calculated_total - invoice_data["total"]) > 0.01:
        errors.append("Line totals don't match invoice total")
    
    return errors

Security Best Practices

API Key Management

import os
from functools import lru_cache

@lru_cache()
def get_api_key():
    """Load API key from secure source."""
    # Environment variable (good for containers)
    if os.environ.get("GOROUTE_API_KEY"):
        return os.environ["GOROUTE_API_KEY"]
    
    # AWS Secrets Manager (production)
    import boto3
    client = boto3.client("secretsmanager")
    secret = client.get_secret_value(SecretId="goroute/api-key")
    return secret["SecretString"]

Webhook Security

import hmac
import hashlib

def verify_webhook(payload: bytes, signature: str, secret: str) -> bool:
    """Verify GoRoute webhook signature."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(f"sha256={expected}", signature)

Audit Logging

import logging
from datetime import datetime

audit_logger = logging.getLogger("audit")

def log_transaction(action, transaction_id, details):
    """Log all Peppol transactions for audit trail."""
    audit_logger.info({
        "timestamp": datetime.utcnow().isoformat(),
        "action": action,
        "transaction_id": transaction_id,
        "user": get_current_user(),
        "details": details
    })

# Usage
log_transaction("send", tx_id, {"receiver": "0192:123456789", "amount": 1000})

Performance Optimization

Batch Operations

# Instead of individual calls
for invoice in invoices:
    send_invoice(invoice)  # Slow!

# Use the batch API: create the batch, then process it
batch = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/batches",
    headers={"X-API-Key": api_key, "Content-Type": "application/json"},
    json={"name": "January invoices", "invoices": invoices, "batch_type": "send"}
).json()

requests.post(
    f"https://app.goroute.ai/peppol-api/api/v1/batches/{batch['id']}/process",
    headers={"X-API-Key": api_key}
)

Batching is two calls, not one: creating a batch stages the invoices and returns 201 with the batch id and a progress object, and a separate process call submits them. See Batch Sending for the batch size limit, the validate-only dry run, and the status model.

Connection Pooling

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Create session with retry logic
session = requests.Session()
retries = Retry(total=3, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retries, pool_connections=10, pool_maxsize=20)
session.mount("https://", adapter)

# Reuse session
response = session.post(url, json=data)

Caching Lookups

import redis
import json

r = redis.Redis()

def lookup_participant(scheme: str, identifier: str):
    """Lookup with caching."""
    cache_key = f"peppol:{scheme}:{identifier}"
    
    # Check cache
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # API call
    response = requests.get(
        "https://app.goroute.ai/peppol-api/api/v1/participants/lookup",
        params={"scheme": scheme, "identifier": identifier},
        headers={"X-API-Key": api_key}
    )
    result = response.json()
    
    # Cache for 1 hour (SMP data changes infrequently)
    r.setex(cache_key, 3600, json.dumps(result))
    
    return result

Testing Strategy

Test Environment

import os

class Config:
    # Use test environment for development
    if os.environ.get("ENVIRONMENT") == "production":
        API_BASE = "https://app.goroute.ai/peppol-api"
        SMP_NETWORK = "production"
    else:
        API_BASE = "https://app.goroute.ai/peppol-api"  # Same API
        SMP_NETWORK = "test"  # Different Peppol network

Mock for Unit Tests

import responses

@responses.activate
def test_send_invoice():
    # Mock API response
    responses.add(
        responses.POST,
        "https://app.goroute.ai/peppol-api/api/v1/documents",
        json={
            "transaction_id": "550e8400-e29b-41d4-a716-446655440000",
            "status": "queued",
            "created_at": "2024-01-15T10:30:00Z"
        },
        status=202
    )
    
    result = send_invoice(sample_invoice)
    
    assert result["status"] == "queued"
    assert len(responses.calls) == 1

Quick Reference

Practice	Description
Validate first	Always validate before sending
Retry transients	Retry 429, 5xx errors with backoff
Use idempotency	Prevent duplicates on retry
Track status	Monitor until delivered/failed
Secure keys	Never commit API keys to code
Batch when possible	Use batch API for multiple docs
Cache lookups	Cache SMP lookups (1 hour TTL)
Log everything	Audit trail for compliance

Core Principles​

1. Validate Before Sending​

2. Handle Errors Gracefully​

3. Use Idempotency​

4. Monitor Delivery Status​

Integration Architecture​

Recommended Flow​

Queue-Based Architecture​

Data Quality​

Clean Input Data​

Validate Before Queuing​

Security Best Practices​

API Key Management​

Webhook Security​

Audit Logging​

Performance Optimization​

Batch Operations​

Connection Pooling​

Caching Lookups​

Testing Strategy​

Test Environment​

Mock for Unit Tests​

Quick Reference​

Next Steps​