Python-SDK
Ein Server-Client für Python-Backends, ohne irgendetwas außerhalb der Standardbibliothek. Jeder Aufruf benennt den Nutzer, denn ein Prozess bedient viele davon.
Wo es veröffentlicht ist
prodantix-sdk on PyPI. Requires Python 3.9 or newer.
Installation
pip install prodantix-sdkJeder Aufruf benennt den Nutzer
Ein Browser-Client kann sich merken, wer ihn benutzt; ein Server kann das nicht. Deshalb ist die Distinct-ID das erste Argument von capture, identify, group und den Posteingangs-Abfragen. Das ist der einzige Formunterschied zu den Browser- und Mobil-Clients.
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")Entfernte und lokale Flags
Eine entfernte Prüfung fragt den Server und ist maßgeblich. Eine lokale Prüfung wertet im Prozess eine zwischengespeicherte Momentaufnahme des Regelwerks aus, ohne Hin und Her pro Aufruf, und genau das braucht ein heißer Pfad.
# 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)Zustellung und Herunterfahren
Ereignisse landen in einer threadsicheren Warteschlange, und ein Hintergrundsender schickt sie in Stapeln, nach Anzahl oder nach Zeit, je nachdem was zuerst eintritt. Nutzen Sie den Client als Kontextmanager, dann geht der letzte Stapel beim Verlassen des Blocks hinaus.
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 |