truesight-sdk

Server-side Python SDK for TrueSight analytics ingestion.

Designed for backends that emit events on behalf of authenticated users (Django, Flask, FastAPI, Airflow DAGs, management commands). Sync by design — if you need async, wrap calls in your own task queue.

Installation

pip install truesight-sdk

Python 3.10+.

Quick Start

Single event

from truesight_sdk import TrueSightClient

client = TrueSightClient(
    api_key="ts_server_live_...",
    base_url="https://ingest.truesight.example.com",
)

client.track(
    event_name="Purchased Lite Pack",
    user_id=str(customer.pk),
    properties={"plan_slug": "5-30-days", "amount": 1200},
)

User profile update (identify)

client.identify(
    user_id=str(customer.pk),
    email=customer.email,
    properties={
        "home_locality": "Andheri",
        "favourite_route": "M1",
        "weekly_subscription_active": True,
    },
)

This upserts the user's row in truesight.user_profiles and also appends a $identify event to the stream for history.

Identify merges into the profile — it does not replace it. The server upserts user_profiles field-wise: any field you omit (email, name, mobile_number, or any key in properties) is left untouched, and properties is merged key-wise (new values win). So you can send just the fields you want to change. To explicitly clear a value, list it in a $unset array inside properties, e.g. properties={"$unset": ["category"]} (promoted fields match case-insensitively; custom property keys are matched in their canonical snake_case form, see Naming below).

Erasing a profile. properties={"$unset": "$all"} clears every value on the profile — email, name, mobile number and every trait — in one identify. It exists for account deletion, where the caller wants everything gone and cannot name the traits: a profile carries traits written by four SDKs plus your backend, so any hard-coded list stops covering whatever ships next.

It empties the row; it does not delete it. What survives is user_uid, first_seen, last_seen and event_count — counts and timestamps rather than identity. user_uid is whatever you passed as user_id, so if your identity is itself personal data (an email, a phone number) the surviving key still identifies the person and $all is not sufficient on its own — key your profiles on an opaque id instead. Events already written keep whatever they carried and age out under the events TTL.

Naming. Event names, property keys and trait keys are stored in snake_case; the SDK canonicalises them in as_payload() with the same rule the server uses (truesight_sdk.to_snake_case): "Purchased Lite Pack" → purchased_lite_pack, "cartValue" → cart_value, "Home Hub" → home_hub. A leading $ is kept, values are never touched. Rewritten names are logged at DEBUG on the truesight_sdk logger. Prefer canonical names in new code.

Bulk profile updates (identify_bulk)

For a one-shot backfill — e.g. "set the email on every customer that's missing one" — use identify_bulk. It takes any iterable of IdentifyEvent, streams it lazily (only chunk_size ops in memory at a time), and auto-chunks to the server's 500-per-call cap. Because the server merges, you only send the fields you're changing:

from truesight_sdk import IdentifyEvent, TrueSightClient

client = TrueSightClient(api_key="ts_server_live_...", base_url="...")

# Fill in just the email — name/mobile/properties already on each profile are preserved.
ops = (
    IdentifyEvent(
        user_id=str(c.pk),
        email=c.email,
        event_id=stable_uuid_for(c),   # optional: makes re-runs idempotent
    )
    for c in customers.iterator()      # a generator/QuerySet streams; nothing is buffered whole
)
sent = client.identify_bulk(ops)       # returns the number of profiles sent

identify_bulk is not transactional — chunks POST sequentially, so if one chunk fails the earlier ones are already applied. Pass a stable event_id per op so a re-run safely converges (retries collapse to the same row).

For continuous, interleaved bulk emission (long-running jobs that mix track and identify), prefer BatchingClient below; for a finite list to push once, identify_bulk is simpler.

Batched ingestion (Airflow / cron / bulk syncs)

For workloads that emit many events in a tight loop, use BatchingClient to amortize HTTP overhead:

from truesight_sdk import BatchingClient, TrueSightClient

inner = TrueSightClient(api_key="ts_server_live_...", base_url="...")

with BatchingClient(inner, batch_size=100, flush_interval_seconds=5.0) as batcher:
    for customer in qs.iterator():
        batcher.identify(
            user_id=str(customer.pk),
            properties=build_profile(customer),
        )
# Buffers drain on context exit.

BatchingClient is thread-safe; multiple producer threads can call track() / identify() concurrently.

Reading events (admin queries)

The SDK also wraps TrueSight's admin event-query endpoint for read-path consumers:

from truesight_sdk import AdminQueryClient

reader = AdminQueryClient(admin_token="...", base_url="https://admin.truesight.example.com")

page = reader.fetch_events(
    project_id="b219fb11-9a63-4843-8126-a3dc05b330a5",
    event_name="Purchased Lite Pack",
    from_=datetime(2026, 5, 1, tzinfo=timezone.utc),
    to=datetime(2026, 5, 28, tzinfo=timezone.utc),
    limit=500,
)

# Fetch a single user's profile + stats (email, name, properties,
# first_seen, last_seen, event_count). Raises `NotFound` (404) if the
# user_uid has never been seen by ingest.
profile = reader.fetch_profile(
    project_id="b219fb11-9a63-4843-8126-a3dc05b330a5",
    user_uid="42",
)

Feature flags and experiments (≥ 0.6.0)

FlagsClient evaluates flags locally. A background thread polls GET /v1/flags/definitions (ETag-conditional, every poll_interval seconds) and the id lists those definitions reference. Each get_variant call is pure CPU against the in-memory snapshot, with no network on the request path.

from truesight_sdk import BatchingClient, FlagsClient, TrueSightClient

flags = FlagsClient(
    api_key="ts_server_live_...",           # server-scoped key
    base_url="https://ts-ingest.cityflo.net",
    poll_interval=30,
    exposure_client=BatchingClient(TrueSightClient(api_key="ts_server_live_...",
                                                   base_url="https://ts-ingest.cityflo.net")),
)

variant = flags.get_variant(
    "home_dock_v1",
    {"user_id": str(customer.user_id)},     # the TrueSight user id (Django User.pk)
    attributes={"city": "mumbai", "platform": "android", "app_version": "5.12.0"},
    default="control",
)
price_cfg = flags.get_payload("pack_price_test", {"user_id": uid}, default={})
if flags.is_enabled("new_refund_flow", {"user_id": uid}):
    ...

• Never raises, never blocks. Before the first load, or on any error, the caller's default comes back. evaluate() returns the full Evaluation, where reason is not_ready, unknown_flag or error in those cases.
• Exposure logging is automatic. When an evaluation places the unit in a running experiment, one $experiment_exposure event is sent through exposure_client per (flag, unit, variant, iteration). An in-memory LRU holds 100k entries for 6 h. Pass track_exposure=False for reads that don't expose the user to the variant (for example roster draws). evaluate_all() never logs exposures.
• Server-sent exposures need a user_id. Device- or anonymous-unit exposures must come from the client SDKs.
• Fork-safe. The client can be created at import time under gunicorn --preload or Celery prefork. Each child restarts its own poller.
• Warm boot. Pass cache= (any object with get(key) / set(key, value), such as Django's cache) so a fresh worker serves the last definitions before its first poll completes.

Django

# common/experiments.py
from django.core.cache import cache
from django.conf import settings
from truesight_sdk import BatchingClient, FlagsClient, TrueSightClient

_flags = FlagsClient(
    api_key=settings.TRUESIGHT_SERVER_API_KEY,
    base_url=settings.TRUESIGHT_BASE_URL,
    exposure_client=BatchingClient(TrueSightClient(settings.TRUESIGHT_SERVER_API_KEY,
                                                   settings.TRUESIGHT_BASE_URL)),
    cache=cache,
)

def get_variant(key, customer, default, **attributes):
    return _flags.get_variant(key, {"user_id": str(customer.user_id)}, attributes, default)

API Keys

Server keys are scoped — they can only call /v1/server/* endpoints. Issue one via the CLI:

truesight projects api-keys create --scope server --label "<your service>"

The plaintext key is returned once at creation time. Store it in your secrets manager.

Error Handling

All errors subclass TrueSightError:

from truesight_sdk import (
    AuthError,        # 401 — bad / revoked key
    Forbidden,        # 403 — wrong scope for this endpoint
    ValidationError,  # 400/422 — payload rejected
    RateLimited,      # 429 — slow down (after SDK's own retry budget)
    ServerError,      # 5xx — TrueSight is degraded (after retries)
    TrueSightError,   # base class — catch this to handle any SDK error
)

try:
    client.track("x", user_id="42")
except RateLimited:
    schedule_retry(...)
except TrueSightError as exc:
    log.exception("truesight ingest failed", request_id=exc.request_id)

Every error carries status_code, request_id (when the server returned one), and response_body for log correlation.

Retry Idempotency

The SDK auto-generates a fresh event_id (UUIDv4) for every event when one isn't supplied. If you need retry idempotency (e.g. inside a Celery task that may run twice), pass an explicit event_id stable across retries:

from truesight_sdk import TrackEvent
from uuid import uuid5, NAMESPACE_URL

stable_id = uuid5(NAMESPACE_URL, f"booking-confirmed:{booking.pk}")
client.track_batch([
    TrackEvent(
        event_name="Booking Confirmed",
        user_id=str(booking.customer_id),
        event_id=stable_id,
        properties={"booking_id": booking.pk},
    ),
])

The server dedups on (project_id, event_id) via ClickHouse's ReplacingMergeTree, so retries with the same id collapse to a single row.

Attributing a server event to a client platform

Server-emitted events land under platform = "server" unless you say which client the action came from. For transactional events the backend owns (bookings, purchases, cancellations) pass the platform of the originating request so dashboards still split them by platform:

TrackEvent(
    event_name="completed_booking",
    user_id=str(user.pk),
    event_id=stable_id,
    platform=request.headers.get("Platform-Name"),  # "Apple" / "Android" / None
    app_version=request.headers.get("App-Version"),
    properties={...},
)

The server canonicalises the value ("Apple" → "ios", "Android" → "android", blank → "server"); os_name defaults to match. source stays "server", so "which platform" and "which producer" remain separate columns.

Passive events (≥ 0.7.0)

TrueSight's active-user numbers (DAU / WAU / MAU, new users, Device Mix) count a user only for events that are not passive. Pass passive=True for events the backend emits about a user who did nothing — a push / email / scorecard it sent, a scheduled verdict, an enrolment — so a nightly send to every rider does not become a day of "active users":

client.track("sure_bundle_scorecard_sent", user_id=str(user.pk), passive=True)

The SDK sends it as properties["$passive"] = True; the event is stored and queryable as usual. $experiment_exposure is always passive. Do not mark real user actions that happen to be emitted server-side (a booking, a purchase, a cancellation): the user did those.

Development

# Install in dev mode
pip install -e ".[dev]"

# Run unit tests (no network)
pytest

# Run integration tests against a local TrueSight stack
export TRUESIGHT_BASE_URL=http://localhost:8080
export TRUESIGHT_API_KEY=ts_server_test_...
pytest -m integration

# Lint + typecheck
ruff check src/ tests/
mypy src/

Versioning

Semver. 0.x is pre-1.0 — the public API may shift between minor versions; pin the minor version until 1.0.

License

MIT — see top-level repo.