truesight-sdk 0.2.0

Server-side Python SDK for TrueSight analytics ingestion

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 (latest-write-wins merge of the properties blob) and also appends a $identify event to the stream for history.

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",
)

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.

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.