segops-admin 0.2.2

Server-side Python Admin SDK for the SegOps platform — programmatic access to segments, products, schemas, activations, queries, pages, and analytics.

segops_admin

Server-side Python Admin SDK for the SegOps platform. Programmatic access to segments, products, schemas, activations, AI queries, landing pages, analytics, and more — the same surface the dashboard uses.

Server-side only. Requires a secret API key (sk_…). Never expose secret keys to browsers or mobile apps.

Zero required runtime dependencies (stdlib only). Python 3.9+.

Install

pip install segops_admin

Quick Start

from segops_admin import SegOpsAdmin

client = SegOpsAdmin(api_key="sk_...")  # or set SEGOPS_API_KEY env var

# List all segments (paginated)
page = client.segments.list()
print(page["results"])

# Trigger a recompute
client.segments.recompute(12)

# Iterate all segments across pages
for segment in client.segments.iterate():
    print(segment["name"])

Resources

Every resource exposes the same CRUD surface plus resource-specific actions:

Namespace	Path prefix	Notable actions
segments	/segments/	recompute, members, membership, export, preview, estimate, versions
product_segments	/product-segments/	recompute
products	/pim/products/	bulk_upsert, score, score_all, facets, schema, import_csv, import_status, confirm_import
schemas	/schemas/	infer, validate
activations	/activations/	sync
users	/explorer/users/	search, get, events, segments, explain, publish_sandbox
queries	/ai-reach/queries/	run, results, simulate, dashboard, win_rate
pages	/ai-pages/pages/	bulk_generate, regenerate, build, publish, export, query
analytics	/analytics/	overview, segments, overlap, activations, usage (read-only)

CRUD methods (all resources except users and analytics)

client.segments.list({"page": 1, "page_size": 20})  # paginated dict
client.segments.iterate()                             # generator over all items
client.segments.get(id)
client.segments.create({"name": "VIP", "definition": {...}})
client.segments.update(id, {"name": "VIP v2"})
client.segments.delete(id)

Segments

client.segments.preview(definition)          # estimate without persisting
client.segments.estimate(id)
client.segments.versions(id)
client.segments.recompute(id)                # POST /segments/{id}/compute/
client.segments.members(id, {"page": 1})
client.segments.membership("user-123")
client.segments.export(id, format="csv")     # or "parquet"

Products (PIM)

client.products.bulk_upsert([{"sku": "SKU-1", "name": "Alpha"}, ...])
client.products.import_csv(open("products.csv", "rb").read(), "products.csv")
client.products.import_status(job_id)
client.products.confirm_import(job_id, {"confirmed": True})
client.products.score(id)
client.products.score_all()
client.products.facets()
client.products.schema()

Users (Explorer)

client.users.search({"q": "alice"})
client.users.get("user-123")
client.users.events("user-123", {"since": "2024-01-01"})
client.users.segments("user-123")
client.users.explain("user-123", segment_id=5)
client.users.publish_sandbox({"user_id": "u1", "event_type": "page_viewed", "payload": {}})

AI Queries

client.queries.run(id)
client.queries.results(id)
client.queries.simulate({"query_text": "best running shoes"})
client.queries.dashboard()
client.queries.win_rate()

Pages

client.pages.bulk_generate({"segment_id": 3, ...})
client.pages.regenerate(id)
client.pages.build(id, {"template": "default"})
client.pages.publish(id)
client.pages.export(id, format="html")   # "tsx" | "html" | "json"
client.pages.query("running shoes")

Pagination

# Single page (dict with count / next / previous / results)
page = client.segments.list({"page": 1, "page_size": 50})
print(page["count"])

# Iterate all items transparently
for segment in client.segments.iterate():
    process(segment)

Errors

from segops_admin import SegOpsApiError

try:
    client.segments.get(9999)
except SegOpsApiError as exc:
    print(exc.status)   # 404
    print(exc.body)     # {"detail": "Not found."}

Idempotency

All mutating operations (create, recompute, export, etc.) automatically send a unique Idempotency-Key header. This is forward-compatible with the upcoming server-side idempotency storage (24h replay window) — the SDK sends the header today even before the server honours it.

Retries

The client retries on 429 and 5xx responses with exponential backoff (default 3 retries, cap 30s). Set max_retries=0 to disable.

client = SegOpsAdmin(api_key="sk_...", max_retries=5)

Webhook Verification

from segops_admin.webhook import verify_webhook

raw_body = request.get_data()
sig = request.headers.get("X-SegOps-Signature", "")
if not verify_webhook(raw_body, sig, os.environ["SEGOPS_WEBHOOK_SECRET"]):
    abort(401)

SegOps signs the raw request body with HMAC-SHA256 and sends X-SegOps-Signature: sha256=<hex>.

CLI

# Install exposes the `segops-admin` command.
pip install segops_admin

export SEGOPS_API_KEY=sk_...
export SEGOPS_API_URL=https://api.segops.ai/api  # optional

segops-admin segments list
segops-admin segments recompute 12
segops-admin products import products.csv
segops-admin pages export 8 --format=html
segops-admin --help

All output is JSON; exit code 0 on success, 1 on API error, 2 on usage error.

Configuration

SegOpsAdmin(
    api_key="sk_...",
    base_url="https://api.segops.ai/api",  # optional override
    max_retries=3,                          # default 3
)

License

MIT