SDK serveur Python

keenauth-python est le SDK serveur pour Python 3.10 ou une version ultérieure. Il appelle l'API de gestion avec des méthodes typées et ajoute le comportement commun à tous les SDK serveur KeenAuth : authentification avec une clé d'API ou un jeton machine-à-machine private_key_jwt, nouvelles tentatives, pagination par curseur, erreurs typées, vérification des signatures de webhooks et avertissements de dépréciation. Le paquet n'est pas encore publié sur PyPI.

Pour la connexion dans une application web, voir Django, Flask et FastAPI. Pour protéger une API qui reçoit des jetons porteurs, voir Protéger votre API.

Installation

pip install keenauth-python

Le signataire private_key_jwt nécessite le paquet cryptography. Installez-le avec l'extra m2m :

pip install "keenauth-python[m2m]"

Appeler l'API de gestion avec une clé d'API

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)

Chaque classe d'API générée est disponible en tant qu'attribut : client.users, client.organizations, client.api_keys et ainsi de suite. L'URL de base doit utiliser https ; http n'est accepté que pour localhost. Le SDK ne suit jamais une redirection, afin que votre identifiant ne soit jamais envoyé à un autre hôte.

Utiliser un jeton de gestion private_key_jwt

Une application de gestion s'authentifie avec sa clé privée au lieu d'une clé d'API. Le SDK signe une assertion client (RFC 7523), obtient un jeton d'API de gestion auprès du point de terminaison de jetons de votre environnement et le renouvelle 60 secondes avant son expiration :

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)

Nouvelles tentatives

Le SDK relance une requête qui a échoué avec 429, 502, 503 ou 504, ou avec une erreur de connexion ou un délai dépassé. Il ne relance que les requêtes GET, PUT, DELETE et POST qui portent un Idempotency-Key. Un PATCH qui porte If-Match n'est relancé que sur 429 : le limiteur de débit refuse une requête avant son exécution, et If-Match empêche une modification d'être appliquée deux fois. Tout autre PATCH n'est jamais relancé. Il effectue au maximum 3 tentatives au total.

Pagination

Les méthodes de liste renvoient une page : data et list_metadata avec les curseurs after et before. paginate parcourt chaque page et ne récupère la page suivante que lorsque vous l'atteignez :

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

Erreurs

Une réponse d'erreur en application/problem+json (RFC 9457) lève keenauth.ProblemError. Utilisez ses champs code, status, detail, request_id et errors. Un code que cette version du SDK ne connaît pas est tout de même renvoyé sous forme de chaîne, de sorte que les nouveaux codes d'erreur du serveur ne cassent pas votre code.

from keenauth import ProblemError

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

Vérifier les webhooks

KeenAuth signe les webhooks avec le schéma Standard Webhooks. WebhookVerifier vérifie les en-têtes webhook-id, webhook-timestamp et webhook-signature à l'aide du secret idp_whsec_ de votre point de terminaison :

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

Événements et revendications typés

keenauth.events contient une dataclass gelée pour chaque type d'événement client, générée à partir des schémas publics d'événements. parse_event vérifie un corps déjà validé et renvoie la classe de son type. keenauth.webhooks.verify_event vérifie et analyse en un seul appel. Affinez avec match ou 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 type les revendications d'un jeton vérifié : parse_id_token_claims (OpenID Connect Core) et parse_access_token_claims (RFC 9068). Elles vérifient les types des revendications mais ne vérifient rien ; appelez-les donc sur les claims renvoyés par le vérificateur :

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 est toujours un tuple. Une revendication incompatible lève ClaimsParseError avec son nom.

Avertissements de dépréciation

Lorsqu'une réponse contient un en-tête Deprecation (RFC 9745) ou Sunset (RFC 8594), le SDK écrit un avertissement par processus dans le journal keenauth avec le module standard logging. L'avertissement indique la méthode, la ressource et les valeurs des en-têtes ; il ne contient jamais d'identifiants, de chaînes de requête ni d'identifiants d'accès. La requête elle-même n'est pas affectée.

import logging

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

En-tête de télémétrie

Chaque requête porte l'en-tête Idp-Client: keenauth-python/<version> (python/<version>). Il ne contient aucune donnée personnelle et ne peut pas être désactivé ; KeenAuth l'utilise pour le support et les avis de dépréciation.