Python サーバー SDK

keenauth-python は Python 3.10 以降向けのサーバー SDK です。型付きメソッドで管理 API を呼び出し、すべての KeenAuth サーバー SDK に共通する機能を提供します。具体的には、API キーまたは private_key_jwt のマシン間 (M2M) トークンによる認証、リトライ、カーソルによるページネーション、型付きエラー、Webhook の署名検証、非推奨警告です。このパッケージはまだ PyPI に公開されていません。

Web アプリでのサインインについては、Django、Flask、FastAPI を参照してください。ベアラートークンを受け取る API を保護するには、API の保護 を参照してください。

インストール

pip install keenauth-python

private_key_jwt の署名には cryptography パッケージが必要です。m2m エクストラを指定してインストールします:

pip install "keenauth-python[m2m]"

API キーで管理 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)

生成されたすべての API クラスは属性として利用できます: client.users、client.organizations、client.api_keys など。ベース URL は https を使用する必要があり、平文の http は localhost の場合のみ許可されます。SDK はリダイレクトを追跡しないため、認証情報が別のホストに送信されることはありません。

private_key_jwt の管理トークンを使用する

管理アプリケーションは API キーの代わりに秘密鍵で認証します。SDK はクライアントアサーション (RFC 7523) に署名し、環境のトークンエンドポイントから管理 API トークンを取得し、有効期限の 60 秒前に更新します:

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)

リトライ

SDK は 429、502、503、504 で失敗したリクエスト、または接続エラーやタイムアウトになったリクエストをリトライします。リトライするのは GET、PUT、DELETE と、Idempotency-Key を含む POST リクエストのみです。If-Match を含む PATCH は 429 の場合のみリトライされます。レート制限は処理前にリクエストを拒否するため、また If-Match により変更が二重に適用されることを防ぐためです。それ以外の PATCH はリトライされません。合計で最大 3 回試行します。

ページネーション

一覧メソッドは 1 ページを返します。data と、after および before カーソルを含む list_metadata です。paginate はすべてのページを順にたどり、必要になったときにのみ次のページを取得します:

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

エラー

application/problem+json (RFC 9457) 形式のエラーレスポンスは keenauth.ProblemError を送出します。code、status、detail、request_id、errors フィールドを使用できます。この SDK バージョンが認識しない 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)

Webhook を検証する

KeenAuth は Standard Webhooks 方式で Webhook に署名します。WebhookVerifier は webhook-id、webhook-timestamp、webhook-signature ヘッダーを、エンドポイントの idp_whsec_ シークレットと照合して検証します:

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

型付きイベントとクレーム

keenauth.events には、公開されているイベントスキーマから生成された、すべての顧客向けイベントタイプごとの frozen dataclass があります。parse_event は検証済みのボディを検査し、そのタイプに対応するクラスを返します。keenauth.webhooks.verify_event は検証と解析を 1 回の呼び出しで行います。match または 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 は、検証済みトークンのクレームに型を付けます: parse_id_token_claims (OpenID Connect Core) と parse_access_token_claims (RFC 9068) です。これらはクレームの型を検査しますが、検証は行いません。検証器が返す claims に対して呼び出してください:

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 は常にタプルです。対応しないクレームがあると、そのクレーム名を含む ClaimsParseError が送出されます。

非推奨警告

レスポンスに Deprecation (RFC 9745) または Sunset (RFC 8594) ヘッダーがある場合、SDK は標準の logging モジュールを使い、プロセスごとに 1 回、keenauth ロガーへ警告を出力します。警告にはメソッド、リソース、ヘッダーの値が含まれます。ID、クエリ文字列、認証情報が含まれることはありません。リクエスト自体には影響しません。

import logging

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

テレメトリヘッダー

すべてのリクエストには Idp-Client: keenauth-python/<version> (python/<version>) が含まれます。個人データは含まれず、無効にすることはできません。KeenAuth はサポートおよび非推奨の通知に使用します。