segops 0.1.0

Behavioral segmentation ingestion SDK for Python — thread-safe event client with batching and the public-key session handshake.

segops

Behavioral segmentation ingestion SDK for Python. Thread-safe event client with built-in batching, the public-key session handshake, and graceful shutdown. Zero required runtime dependencies (stdlib only).

Install

pip install segops

Requires Python 3.9+.

Usage

Server — secret key (sk_…)

A secret key authenticates directly and must never reach a browser or mobile app.

import segops

client = segops.SegOpsClient("https://api.segops.ai", "sk_...")

client.track(user_id="u-123", event_type="order_placed", payload={"total": 42})
client.shutdown()  # flush + stop the background thread on exit

Or use it as a context manager (flushes on exit):

with segops.SegOpsClient("https://api.segops.ai", "sk_...") as client:
    client.track(user_id="u-123", event_type="page_viewed", payload={"path": "/home"})

Public key (pk_…) — session handshake

A public key can't write directly; the SDK exchanges it for a short-lived session token via the handshake. Provide get_user to return the current user context. To bind events to a logged-in user, sign the user id on your backend (see below) and return it from get_user.

import segops

client = segops.SegOpsClient(
    "https://api.segops.ai",
    "pk_...",
    get_user=lambda: segops.UserContext(
        user_id=current_user.id,
        user_id_sig=signed["user_id_sig"],  # from your backend
        user_id_ts=signed["user_id_ts"],
    ),
)

The token is cached, refreshed 60s before expiry, and re-minted on a 401.

Identify

client.identify(
    user_id="u-123",
    traits={"email": "[e-mail]", "plan": "pro"},
)

Emits a context_identified event whose payload is the traits.

Options

SegOpsClient(api_url, api_key, *, get_user=None, batch=True, batch_size=20, flush_interval=5.0, on_error=None)

Argument	Default	Description
api_url	—	Base URL of your SegOps deployment
api_key	—	sk_… (server) or pk_… (handshake)
get_user	—	Required for pk_…: returns a UserContext at mint time
batch	True	Buffer events and flush automatically; False sends each immediately
batch_size	20	Max events before an automatic flush
flush_interval	5.0	Periodic flush interval (seconds)
on_error	log to stderr	Called when a background flush fails

Signing a user id (backend only)

Run this on your backend only — never ship the HMAC secret to a client.

import os
import segops

signed = segops.sign_user_id(os.environ["SEGOPS_HMAC_SECRET"], current_user.id)
# -> {"user_id": ..., "user_id_sig": ..., "user_id_ts": ...} — hand to get_user()

Enable "require signed user_id" on the public key so the server rejects unsigned or forged identities.

License

MIT