Skip to main content

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/validate",
    headers={"X-API-Key": api_key},
    data=invoice_xml
)

if validation.json()["valid"]:
    # Then send
    response = requests.post(
        "https://app.goroute.ai/peppol-api/api/v1/send",
        headers={"X-API-Key": api_key},
        json=payload
    )

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

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/send",
    headers={
        "X-API-Key": api_key,
        "Idempotency-Key": idempotency_key
    },
    json=payload
)

# Same key = same result if retried

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​

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 batch API
batch_response = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/batch",
    headers={"X-API-Key": api_key},
    json={"documents": invoices}
)

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/send",
        json={"transaction_id": "tx_123", "status": "accepted"},
        status=200
    )
    
    result = send_invoice(sample_invoice)
    
    assert result["status"] == "accepted"
    assert len(responses.calls) == 1

Quick Reference​

PracticeDescription
Validate firstAlways validate before sending
Retry transientsRetry 429, 5xx errors with backoff
Use idempotencyPrevent duplicates on retry
Track statusMonitor until delivered/failed
Secure keysNever commit API keys to code
Batch when possibleUse batch API for multiple docs
Cache lookupsCache SMP lookups (1 hour TTL)
Log everythingAudit trail for compliance

Next Steps​