Dativery Peppoš

Ukázka: saas_verified_onboarding_example.py

Tento skript demonstruje zkrácený ověřený postup onboardingu: autentizace, vytvoření účastníka označeného jako ověřený a předregistrovaného v Peppol, registrace webhooků a odeslání XML dokumentu.

Co tato ukázka předvádí

Autentizace pomocí klientských přihlašovacích údajů pro získání přístupového tokenu.
Vytvoření účastníka s příznaky mark-as-verified a register-to-peppol v dotazu.
Registrace webhooku pro události doručení a příchozích dokumentů.
Závěrečné volání odeslání dokumentu pomocí XML datové části z lokálního souboru.

saas_verified_onboarding_example.py

Naformátovaný a zvýrazněný zdrojový kód Pythonu

Stáhnout soubor

#!/usr/bin/env python3
"""SaaS onboarding example: create participant → verify → register in Peppol → send document."""
import json, os, time, urllib.error, urllib.parse, urllib.request
from pathlib import Path

BASE          = os.getenv("PEPPOS_BASE_URL",              "https://sandbox.pepposh.eu")
CLIENT_ID     = os.getenv("PEPPOS_CLIENT_ID",             "YOUR-API-KEY")
CLIENT_SECRET = os.getenv("PEPPOS_CLIENT_SECRET",         "YOUR-API-SECRET")
WEBHOOK_URL   = os.getenv("PEPPOS_WEBHOOK_URL",           "https://your-saas.example.com/webhooks/peppol")
INVOICE_FILE    = Path(os.getenv("INVOICE_FILE",          "invoice/sample-ubl-invoice.xml"))
PARTICIPANT_ID = os.getenv("PARTICIPANT_ID",              "my-customer-1")

def request(method, url, token=None, body=None, content_type="application/json"):
    headers = {"Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    data = None
    if body is not None:
        if isinstance(body, (dict, list)):
            data, headers["Content-Type"] = json.dumps(body).encode(), "application/json"
        else:
            data, headers["Content-Type"] = str(body).encode(), content_type
    try:
        with urllib.request.urlopen(
            urllib.request.Request(url, data=data, method=method, headers=headers)
        ) as r:
            raw = r.read().decode()
            return r.getcode(), json.loads(raw) if raw.strip() else {}
    except urllib.error.HTTPError as e:
        raw = e.read().decode()
        return e.code, json.loads(raw) if raw.strip() else {}


def ok(label, code, body):
    if code not in (200, 201, 202):
        raise RuntimeError(f"{label} failed ({code}): {body}")
    return body.get("data", body)


def main():
    # 1. Authenticate
    code, body = request("POST", f"{BASE}/api/auth/token",
        body={"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "client_credentials"})
    token = ok("auth", code, body).get("access_token") or body.get("access_token")
    print(f"[1] token obtained")

    # 2. Create participant
    code, body = request("POST", f"{BASE}/api/participants/ext:{PARTICIPANT_ID}?mark-as-verified&register-to-peppol", token=token,
        body={"name": "Example Company s.r.o.", "registrationNumber": "12345678", "vatId": "CZ12345678", "email": "invoice@example.com", "countryCode": "CZ"})
    pid = ok("create participant", code, body).get("id") or ok("create participant", code, body).get("participantId")
    print(f"[2] participantId={pid}")

    # 4. Register webhook
    code, body = request("POST", f"{BASE}/api/participants/ext:{PARTICIPANT_ID}/webhooks", token=token,
        body={"url": WEBHOOK_URL, "events": ["document.received", "document.delivered"]})
    ok("webhooks", code, body)
    print(f"[6] Webhook registered: {WEBHOOK_URL}")

    # 5. Send invoice document
    code, body = request("POST", f"{BASE}/api/participants/ext:{PARTICIPANT_ID}/documents/send?wait-for-result", token=token,
        body=INVOICE_FILE.read_text(), content_type="application/xml")
    data = ok("send", code, body)
    doc_id = data.get("documentId") or data.get("id")
    print(f"[7] Document sent, documentId={doc_id}")


if __name__ == "__main__":
    main()

Dokumentace API

Ukázka: saas_verified_onboarding_example.py