SDK server Python

keenauth-python è l'SDK server per Python 3.10 o successivo. Chiama l'API di gestione con metodi tipizzati e aggiunge il comportamento che tutti gli SDK server di KeenAuth condividono: autenticazione con una chiave API o con un token machine-to-machine private_key_jwt, retry, paginazione con cursore, errori tipizzati, verifica delle firme dei webhook e avvisi di deprecazione. Il pacchetto non è ancora pubblicato su PyPI.

Per accedere in un'app web, vedi Django, Flask e FastAPI. Per proteggere un'API che riceve bearer token, vedi Proteggere la tua API.

Installazione

pip install keenauth-python

Il signer private_key_jwt richiede il pacchetto cryptography. Installalo con l'extra m2m:

pip install "keenauth-python[m2m]"

Chiamare l'API di gestione con una chiave 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)

Ogni classe API generata è disponibile come attributo: client.users, client.organizations, client.api_keys e così via. L'URL di base deve usare https; http semplice è accettato solo per localhost. L'SDK non segue mai un redirect, quindi la tua credenziale non viene mai inviata a un altro host.

Usare un token di gestione private_key_jwt

Un'applicazione di gestione si autentica con la sua chiave privata invece che con una chiave API. L'SDK firma un'asserzione client (RFC 7523), ottiene un token dell'API di gestione dall'endpoint dei token del tuo ambiente e lo rinnova 60 secondi prima della scadenza:

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)

Retry

L'SDK ritenta una richiesta fallita con 429, 502, 503 o 504, oppure con un errore di connessione o un timeout. Ritenta solo richieste GET, PUT, DELETE e POST che includono un Idempotency-Key. Una PATCH che include If-Match viene ritentata solo su 429: il rate limiter rifiuta una richiesta prima che venga eseguita, e If-Match impedisce che una modifica venga applicata due volte. Ogni altra PATCH non viene mai ritentata. Effettua al massimo 3 tentativi in totale.

Paginazione

I metodi di elenco restituiscono una pagina: data e list_metadata con i cursori after e before. paginate scorre tutte le pagine e richiede la pagina successiva solo quando la raggiungi:

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

Errori

Una risposta di errore in application/problem+json (RFC 9457) genera keenauth.ProblemError. Usa i suoi campi code, status, detail, request_id ed errors. Un code che questa versione dell'SDK non conosce viene comunque restituito come stringa, quindi i nuovi codici di errore del server non rompono il tuo codice.

from keenauth import ProblemError

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

Verificare i webhook

KeenAuth firma i webhook con lo schema Standard Webhooks. WebhookVerifier controlla gli header webhook-id, webhook-timestamp e webhook-signature rispetto al segreto idp_whsec_ del tuo endpoint:

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

Eventi e claim tipizzati

keenauth.events contiene una dataclass frozen per ogni tipo di evento cliente, generata dagli schemi pubblici degli eventi. parse_event verifica un corpo già validato e restituisce la classe del suo tipo. keenauth.webhooks.verify_event verifica e analizza in un'unica chiamata. Restringi il tipo con match o 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 tipizza i claim di un token verificato: parse_id_token_claims (OpenID Connect Core) e parse_access_token_claims (RFC 9068). Controllano i tipi dei claim ma non verificano nulla, quindi chiamali sui claims che restituisce il verifier:

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 è sempre una tupla. Un claim non valido genera ClaimsParseError con il suo nome.

Avvisi di deprecazione

Quando una risposta ha un header Deprecation (RFC 9745) o Sunset (RFC 8594), l'SDK scrive un avviso per processo nel logger keenauth con il modulo standard logging. L'avviso indica il metodo, la risorsa e i valori dell'header; non contiene mai ID, query string o credenziali. La richiesta stessa non è influenzata.

import logging

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

Header di telemetria

Ogni richiesta include Idp-Client: keenauth-python/<version> (python/<version>). Non contiene dati personali e non può essere disattivato; KeenAuth lo usa per il supporto e per gli avvisi di deprecazione.