Python SDK
Серверный клиент для бэкендов на Python, без чего-либо за пределами стандартной библиотеки. Каждый вызов называет пользователя, ведь один процесс обслуживает многих.
Где опубликован
prodantix-sdk on PyPI. Requires Python 3.9 or newer.
Установка
pip install prodantix-sdkКаждый вызов называет пользователя
Браузерный клиент может помнить, кто им пользуется; сервер не может. Поэтому идентификатор пользователя это первый аргумент для capture, identify, group и чтения почтового ящика. Это единственное отличие формы от браузерного и мобильного клиентов.
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")Удалённые и локальные флаги
Удалённая проверка спрашивает сервер и является определяющей. Локальная проверка вычисляет кешированный снимок набора правил внутри процесса, без обращения к серверу на каждый вызов, а именно это и нужно горячему пути.
# 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)Отправка и завершение
События попадают в потокобезопасную очередь, а фоновый отправитель шлёт их пачками по счётчику или по таймеру, смотря что наступит раньше. Используйте клиент как менеджер контекста, и последняя пачка уйдёт при выходе из блока.
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 |