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í
saas_onboarding_example.py
Naformátovaný a zvýrazněný zdrojový kód Pythonu
#!/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()