Python SDK
面向 Python 后端的服务端客户端，不依赖标准库以外的任何东西。每次调用都要指明用户，因为一个进程服务着很多人。
发布位置
prodantix-sdk on PyPI. Requires Python 3.9 or newer.
安装
Bash
pip install prodantix-sdk每次调用都指明用户
浏览器客户端能记住是谁在用它，服务端不能。因此用户 id 是 capture、identify、group 以及收件箱读取的第一个参数。这是它与浏览器、移动客户端之间唯一的形态差异。
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 |