Poštár

Dokumentace API

Ukázka: saas_onboarding_example.py

Tento skript ukazuje kompletní SaaS onboarding: přihlášení, vytvoření účastníka, spuštění ověření, registraci v Peppolu, nastavení webhook callbacků a odeslání dokumentu ISDOC.

Co tato ukázka předvádí

  • Vytvoření účastníka se základními údaji o společnosti.
  • Spuštění ověření a následné dotazování až do schválení.
  • Registrace do Peppolu a konfigurace webhooků.
  • Závěrečné volání pro odeslání ISDOC k ověření celé integrační cesty.

saas_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")
CALLBACK      = os.getenv("PEPPOS_VERIFICATION_CALLBACK", "https://your-saas.example.com/verification/done")
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"))
POLL_INTERVAL = int(os.getenv("POLL_INTERVAL_SECONDS",    "5"))
POLL_TIMEOUT  = int(os.getenv("POLL_TIMEOUT_SECONDS",     "300"))


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", 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}")

    # 3. Initiate verification – user must open the URL in a browser
    callback = urllib.parse.quote(CALLBACK, safe="")
    code, body = request("POST", f"{BASE}/api/participants/{pid}/verify/initiate?callback={callback}", token=token)
    data = ok("verify/initiate", code, body)
    verification_url = data.get("verificationUrl") or data.get("url")
    print(f"[3] Open in browser to verify:\n    {verification_url}")

    input("\nPress Enter after completing verification in the browser...")

    # 4. Poll verification status (alternatively, you can use webhook to receive update for verification status)
    print("[4] Polling verification status...")
    deadline = time.time() + POLL_TIMEOUT
    while time.time() < deadline:
        code, body = request("GET", f"{BASE}/api/participants/{pid}/verify/status", token=token)
        state = (body.get("data", body).get("status") or "").lower()
        print(f"    status: {state}")
        if state in ("verified", "completed", "approved"):
            break
        if state in ("failed", "rejected", "cancelled"):
            raise RuntimeError(f"Verification failed: {state}")
        time.sleep(POLL_INTERVAL)
    else:
        raise TimeoutError("Verification timed out")

    # 5. Register in Peppol
    code, body = request("POST", f"{BASE}/api/participants/{pid}/peppol/register", token=token)
    ok("peppol/register", code, body)
    print(f"[5] Registered in Peppol")

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

    # 7. Send invoice document
    code, body = request("POST", f"{BASE}/api/participants/{pid}/documents/send", 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()