Skip to main content

How to integrate a Python service

This guide is for Python integrations: a Django/Flask back-office, a Celery worker, a data-pipeline step, or a one-off migration script. Keyban ships an official Python client (keyban-api-client) that wraps the DPP Passport REST API; this guide uses it. For Loyalty or any other surface, call the REST API directly with httpx or requests — the URL surface is identical to the Node guide.

At a glance​

  1. keyban-api-client (DPP)PassportClient — list, get, create, update passport models.
  2. No webhook helperKeyban doesn't emit outbound webhooks today. Polling only.
  3. Your Python serviceHolds the X-Api-Key. Talks to Keyban over HTTPS.
  4. No Loyalty in clientLoyalty endpoints are not wrapped — call them through `requests`.
The keyban-api-client wraps the DPP Passport API. Loyalty and other surfaces are not covered by the client — call them through plain requests.

1. Install the client​

The Python client targets Python 3.9+. Install it from PyPI:

pip install keyban-api-client
# or
poetry add keyban-api-client

2. Configure the client​

Like every back-office integration, the API key lives in your secret store, not in the codebase.

import os
from keyban_api_client import PassportClient

client = PassportClient(
    base_url=os.environ.get("KEYBAN_API", "https://api.prod.keyban.io"),
    api_key=os.environ["KEYBAN_API_KEY"],
)

Create the key from Organization → API keys in the Admin Console. The supported scopes and the permissions shape are listed in the Permissions reference.

3. List, read, and create passports​

PassportClient exposes synchronous methods only — there is no async variant today. The core four are shown below (list_passports, get_passport, create_passport_item, update_passport_item).

from uuid import UUID
from datetime import date
from keyban_api_client import PassportClient, PassportData, FilterOperator

# List with a filter
response = client.list_passports(
    filters=[FilterOperator(field="modelNumber", operator="eq", value="WAL-LEATHER-2026")],
    current_page=1,
    page_size=10,
)
print(response.total, len(response.data))

# Read one
passport = client.get_passport(passport_id=UUID("550e8400-e29b-41d4-a716-446655440000"))

# Create an item (writes are item-granularity since 3.0)
data = PassportData(name="Heritage Leather Wallet", manufacturing_date=date(2026, 4, 1))
new_item = client.create_passport_item(
    application=UUID("..."),
    network="BaseSepolia",
    item_number="WAL-LEATHER-2026-0001",
    model_number="WAL-LEATHER-2026",
    data=data.model_dump(),
    certified_paths=["name"],
)

# Update an item — resend the FULL data object: PATCH replaces `data`
# wholesale, a partial dict would drop the other fields (here the certified `name`)
updated = PassportData(name="Heritage Leather Wallet", manufacturing_date=date(2026, 5, 1))
client.update_passport_item(
    passport_id=new_item.id,
    data=updated.model_dump(),
)

Need to create more than a handful of passports? Skip the per-row loop and use the bulk import endpoint covered in Section 4 — one HTTP call, asynchronous processing, idempotent on re-upload.

4. Bulk-import a CSV catalogue​

For initial catalogue migrations or periodic ERP/PIM syncs, POST /v1/imports accepts a JSON array of passports and processes them asynchronously. The client doesn't wrap this endpoint — call it through requests. The body shape is the same CreateDppPassportDto discriminated union as the unitary POST /v1/dpp/passports, but the application is passed as a query parameter, not in each item.

import csv
import os
import time
import requests

KEYBAN_API = os.environ.get("KEYBAN_API", "https://api.prod.keyban.io")
APP_ID = os.environ["KEYBAN_APP_ID"]
HEADERS = {"X-Api-Key": os.environ["KEYBAN_API_KEY"]}

# 1. Build the JSON array from the CSV
rows = []
with open("catalogue.csv", newline="") as f:
    for row in csv.DictReader(f):
        rows.append({
            "granularity": "model",
            "network": "BaseSepolia",
            "modelNumber": row["sku"],
            "name": row["name"],
            "data": {"gtin": row["gtin"], "material": row.get("material", "")},
            "certifiedPaths": ["gtin"],
        })

# 2. Submit the upload — synchronous response carries the job envelope
job = requests.post(
    f"{KEYBAN_API}/v1/imports",
    params={"applicationId": APP_ID, "entityName": "DppPassport"},
    json=rows,
    headers=HEADERS,
).json()

job_id = job["id"]
print(f"Job {job_id} queued — {len(rows)} rows")

# 3. Poll the report until every child row is in a terminal state
while True:
    report = requests.get(
        f"{KEYBAN_API}/v1/imports/{job_id}/report",
        params={"currentPage": 1, "pageSize": len(rows)},
        headers=HEADERS,
    ).json()

    terminal = sum(1 for c in report["data"] if c["status"] in ("completed", "failed"))
    if terminal == report["total"]:
        break
    print(f"  {terminal}/{report['total']} done")
    time.sleep(2)

# 4. Surface the failures
for child in report["data"]:
    if child["status"] == "failed":
        print(f"row {child['entityData']['modelNumber']} → {child['error']}")

Key properties of the import endpoint:

  • Idempotent: the unique tuple (application, granularity, modelNumber, batchNumber, itemNumber) triggers an upsert. Re-running the same export is a no-op when nothing changed, deterministic update otherwise.
  • Per-row validation: a malformed row fails its own child job (status: "failed", error: "...") — the rest of the upload keeps processing. The HTTP 200 only means "upload accepted"; correctness is verified through the report.
  • Resumable: a partially-failed import can be re-submitted as-is once the offending rows are fixed.

For Server-Sent Events streaming and the full failure-handling reference, see Bulk import in the PHP guide — same payload shape, the only platform-specific bit is the HTTP client.

5. Handle errors​

keyban-api-client raises a single exception type — KeybanAPIError, a subclass of requests.HTTPError. The structured error body is parsed and exposed on error.detail:

from keyban_api_client import KeybanAPIError

try:
    client.create_passport_item(application=..., network=..., item_number=..., data={}, certified_paths=[])
except KeybanAPIError as exc:
    # exc.status_code is the HTTP status
    # exc.detail is the parsed JSON body (status, title, detail, errors[])
    log.error(
        "Keyban call failed (status %s): %s — %s",
        exc.status_code,
        exc.detail.get("title"),
        exc.detail.get("detail"),
    )
    if exc.status_code == 429:
        time.sleep(int(exc.response.headers.get("Retry-After", "1")))

There are no per-category subclasses (no RateLimitedError, no ValidationError) — branch on status_code and on the fields of the parsed detail. The full strategy is in How to handle Keyban API errors.

6. Loyalty and other surfaces — call REST directly​

The client only wraps DPP Passports. For Loyalty (mint, reward tiers, orders), call the REST API with requests:

import os, requests

KEYBAN_API = os.environ.get("KEYBAN_API", "https://api.prod.keyban.io")
HEADERS = {"X-Api-Key": os.environ["KEYBAN_API_KEY"]}

# Mint loyalty points
r = requests.post(
    f"{KEYBAN_API}/v1/loyalty/account/{account_id}/mint",
    json={"amount": 600},
    headers=HEADERS,
)
r.raise_for_status()
print(r.json())  # {"newamount": 600}

Reference​

MethodPathAuthPurpose
POST/v1/dpp/passportsX-Api-KeyCreate a DPP passport (one row). Wrapped by client.create_passport_item.
GET/v1/dpp/passportsX-Api-KeyList passports. Wrapped by client.list_passports.
POST/v1/importsX-Api-KeyBulk-import DPP passports (async, upsert). Not wrapped — call directly.
GET/v1/imports/:jobId/reportX-Api-KeyPer-row status report for an import job.
POST/v1/loyalty/account/:id/mintX-Api-KeyMint points (call directly — not wrapped).