Python SDK
Python のバックエンド向けサーバークライアントで、標準ライブラリの外には何も使いません。一つのプロセスが多くの利用者を扱うため、呼び出しごとに利用者を指定します。
公開先
prodantix-sdk on PyPI. Requires Python 3.9 or newer.
インストール
Bash
pip install prodantix-sdk呼び出しごとに利用者を指定する
ブラウザのクライアントは誰が使っているかを覚えていられますが、サーバーは覚えられません。そのため capture、identify、group、受信箱の読み取りでは利用者の id が第一引数になります。ブラウザ版やモバイル版との形の違いはこれだけです。
Python
import os
from prodantix import Client
client = Client(
api_key=os.environ["PRODANTIX_PUBLIC_KEY"],
host="https://eu.api.prodantix.com",
)
client.capture("user-123", "order.completed", properties={"total": 42, "currency": "XAF"})
client.identify("user-123", set_props={"plan": "pro"})
client.group("user-123", "company", "acme", properties={"seats": 10})
for message in client.get_inbox("user-123"):
...
client.mark_message_read("user-123", "message-id")リモートフラグとローカルフラグ
リモート判定はサーバーに問い合わせ、これが正となります。ローカル判定はキャッシュしたルールのスナップショットをプロセス内で評価し、呼び出しごとの往復がありません。ホットパスが求めるのはこちらです。
Python
# Remote: asks the server, and is the authoritative answer.
if client.is_feature_enabled("user-123", "new-checkout"):
...
# Local: evaluated in process from a cached snapshot, no round trip per call.
if client.is_feature_enabled_local("user-123", "new-checkout", properties={"plan": "pro"}):
...
every_flag = client.get_all_flags("user-123")
# Every per-flag read records one $feature_flag_called exposure.
variant = client.get_variant("user-123", "new-checkout")
if variant and variant["key"] == "treatment":
...
# A flag targeting a cohort reads the user's memberships from the store;
# pass cohorts=[...] to skip the lookup.
if client.is_feature_enabled_local("user-123", "members", cohorts=["<cohort id>"]):
...
# With stream_flags=True the client holds a Socket.IO subscription and refetches
# the snapshot when a flag changes, instead of waiting for the 30s poll.
client = Client(api_key=os.environ["PRODANTIX_PUBLIC_KEY"], host="https://eu.api.prodantix.com", stream_flags=True)送信と終了
イベントはスレッドセーフなキューに入り、バックグラウンドの送信役が件数かタイマーの早いほうを条件にまとめて送ります。クライアントをコンテキストマネージャとして使えば、ブロックを抜けるときに最後のバッチが送られます。
Python
with Client(api_key=..., host=...) as client:
client.capture("user-123", "page.viewed")
# the final batch is flushed here
# Without the context manager, flush and stop the background flusher yourself.
client.shutdown()| Argument | Default | Purpose |
|---|---|---|
| api_key | required | Public project key (pdx_pub_…) |
| host | required | Ingest base URL |
| flags_host | host | Base URL for the flags and inbox endpoints |
| flush_at | 20 | Queue size that triggers an immediate flush |
| flush_interval_s | 10.0 | Background flush cadence; 0 or less disables the timer |
| max_queue_size | 1000 | Cap; the oldest events are dropped on overflow |
| max_retries | 3 | Delivery retry attempts |
| request_timeout_s | 10.0 | Per-request timeout |
| default_properties | None | A dict or a callable merged into every event |
| on_error | no-op | Called with any capture or background error |
| stream_flags | False | Hold a Socket.IO subscription to flag changes and refetch the snapshot on push |
| socket_factory | stdlib client | Builds the websocket the flag stream uses; the test seam |