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 |