Python-Server-SDK

keenauth-python ist das Server-SDK für Python 3.10 oder neuer. Es ruft die Management-API mit typisierten Methoden auf und bietet das Verhalten, das alle Server-SDKs von KeenAuth gemeinsam haben: Authentifizierung mit einem API-Schlüssel oder einem private_key_jwt-Maschine-zu-Maschine-Token, Wiederholungsversuche, Cursor-Paginierung, typisierte Fehler, Prüfung von Webhook-Signaturen und Warnungen zur Veralterung. Das Paket ist noch nicht auf PyPI veröffentlicht.

Informationen zum Anmelden in einer Web-Anwendung finden Sie unter Django, Flask und FastAPI. Um eine API zu schützen, die Bearer-Token empfängt, lesen Sie Ihre API schützen.

Installieren

pip install keenauth-python

Der private_key_jwt-Signierer benötigt das Paket cryptography. Installieren Sie es mit dem Extra m2m:

pip install "keenauth-python[m2m]"

Management-API mit einem API-Schlüssel aufrufen

import os
from keenauth import KeenAuth

client = KeenAuth("https://api.eu.keenauth.com/v1", os.environ["KEENAUTH_API_KEY"])

page = client.users.list_users(limit=20)
for user in page.data:
    print(user.id)

Jede generierte API-Klasse ist als Attribut verfügbar: client.users, client.organizations, client.api_keys und so weiter. Die Basis-URL muss https verwenden; einfaches http ist nur für localhost zulässig. Das SDK folgt niemals einer Weiterleitung, damit Ihre Anmeldedaten nie an einen anderen Host gesendet werden.

Management-Token mit private_key_jwt verwenden

Eine Management-Anwendung authentifiziert sich mit ihrem privaten Schlüssel statt mit einem API-Schlüssel. Das SDK signiert eine Client-Assertion (RFC 7523), holt damit ein Token für die Management-API vom Token-Endpunkt Ihrer Umgebung und erneuert es 60 Sekunden vor dessen Ablauf:

import os
from keenauth import KeenAuth
from keenauth.m2m import ManagementTokenSource

source = ManagementTokenSource(
    issuer="https://acme.eu.keenauth.app",          # your environment's default host
    client_id="app_01j9zzzzzzzzzzzzzzzzzzzz",       # a management application
    private_key=os.environ["KEENAUTH_PRIVATE_KEY"],  # unencrypted PEM
    resource="https://api.eu.keenauth.com",          # the management API of your region
    scopes=["users:read", "users:write"],            # optional; default: the grant's default scopes
)
client = KeenAuth("https://api.eu.keenauth.com/v1", token_source=source)

Wiederholungsversuche

Das SDK wiederholt eine Anfrage, die mit 429, 502, 503 oder 504 fehlgeschlagen ist, oder bei einem Verbindungsfehler oder Timeout. Wiederholt werden nur GET-, PUT-, DELETE- und POST-Anfragen mit einem Idempotency-Key. Ein PATCH mit If-Match wird nur bei 429 wiederholt: Der Rate-Limiter lehnt eine Anfrage ab, bevor sie ausgeführt wird, und If-Match verhindert, dass eine Änderung zweimal angewendet wird. Jedes andere PATCH wird nie wiederholt. Insgesamt sind höchstens 3 Versuche vorgesehen.

Paginierung

Listenmethoden geben eine Seite zurück: data und list_metadata mit den Cursorn after und before. paginate durchläuft alle Seiten und ruft die nächste Seite erst ab, wenn Sie sie erreichen:

from keenauth import paginate

for user in paginate(client.users.list_users, limit=100, status="active"):
    if user.email == "ada@example.com":
        break  # no further page is requested

Fehler

Eine Fehlerantwort im Format application/problem+json (RFC 9457) löst keenauth.ProblemError aus. Verwenden Sie die Felder code, status, detail, request_id und errors. Ein code, den diese SDK-Version nicht kennt, wird trotzdem als Zeichenkette zurückgegeben, sodass neue Fehlercodes des Servers Ihren Code nicht brechen.

from keenauth import ProblemError

try:
    client.users.get_user("usr_01j9zzzzzzzzzzzzzzzzzzzz")
except ProblemError as e:
    if e.code == "not_found":
        ...
    print(e.request_id)

Webhooks prüfen

KeenAuth signiert Webhooks nach dem Standard-Webhooks-Schema. WebhookVerifier prüft die Header webhook-id, webhook-timestamp und webhook-signature anhand des Geheimnisses idp_whsec_ Ihres Endpunkts:

import os
from flask import Flask, request
from keenauth.webhooks import WebhookVerificationError, WebhookVerifier

app = Flask(__name__)
verifier = WebhookVerifier(os.environ["KEENAUTH_WEBHOOK_SECRET"])

@app.post("/webhooks/keenauth")
def keenauth_webhook():
    try:
        verifier.verify(request.get_data(), request.headers)
    except WebhookVerificationError as e:
        return {"error": e.reason}, 400
    event = request.get_json()
    ...
    return "", 204

Typisierte Ereignisse und Claims

keenauth.events enthält für jeden Ereignistyp des Kunden eine eingefrorene Dataclass, die aus den öffentlichen Ereignisschemas generiert wird. parse_event prüft einen verifizierten Body und gibt die Klasse seines Typs zurück. keenauth.webhooks.verify_event verifiziert und parst in einem Aufruf. Verengen Sie den Typ mit match oder isinstance:

from keenauth.events import EmailSendingSuspendedEvent, UnknownEvent, UserCreatedEvent
from keenauth.webhooks import verify_event

event = verify_event(verifier, request.get_data(), request.headers)
match event:
    case UserCreatedEvent():
        print(event.data.object.id)
    case EmailSendingSuspendedEvent():
        print(event.data.object.reason)
    case UnknownEvent():
        pass  # a type this SDK version does not know yet; event.type and event.raw are still available

keenauth.claims typisiert die Claims eines verifizierten Tokens: parse_id_token_claims (OpenID Connect Core) und parse_access_token_claims (RFC 9068). Sie prüfen die Typen der Claims, verifizieren aber nichts. Rufen Sie sie daher mit den claims auf, die der Verifier zurückgibt:

from keenauth.claims import parse_access_token_claims

at = verifier.verify_access_token(token)
claims = parse_access_token_claims(at.claims)
claims.client_id     # str
claims.org_roles     # tuple[str, ...] | None
claims.custom        # your own claims

aud ist immer ein Tupel. Ein Claim, der nicht passt, löst ClaimsParseError mit seinem Namen aus.

Warnungen zur Veralterung

Wenn eine Antwort einen Header Deprecation (RFC 9745) oder Sunset (RFC 8594) enthält, schreibt das SDK eine Warnung pro Prozess in den Logger keenauth über das Standardmodul logging. Die Warnung nennt die Methode, die Ressource und die Header-Werte; sie enthält niemals IDs, Abfragezeichenfolgen oder Anmeldedaten. Die Anfrage selbst bleibt unverändert.

import logging

logging.getLogger("keenauth").setLevel(logging.WARNING)

Telemetrie-Header

Jede Anfrage enthält Idp-Client: keenauth-python/<version> (python/<version>). Der Header enthält keine personenbezogenen Daten und kann nicht deaktiviert werden; KeenAuth verwendet ihn für Support und Hinweise zur Veralterung.