Python SDK

The telemetry-sh package provides Telemetry for synchronous applications and TelemetryAsync for asyncio services. Both clients send each method call immediately to the Telemetry HTTP API. Keep the API key in server-side configuration.

Install and initialize

python -m pip install telemetry-sh

Synchronous code:

import os
from telemetry_sh import Telemetry

telemetry = Telemetry()
telemetry.init(os.environ.get("TELEMETRY_API_KEY"))

Asynchronous code:

import os
from telemetry_sh import TelemetryAsync

telemetry = TelemetryAsync()
telemetry.init(os.environ.get("TELEMETRY_API_KEY"))

init is a normal method for both clients; do not await it. Initialize once per process with a write-scoped key for event producers or a read-scoped key for query-only jobs.

Send an event synchronously

import time
import uuid

started_at = time.perf_counter()
event_id = str(uuid.uuid4())

try:
    response = telemetry.log("job_completed", {
        "event_id": event_id,
        "job_name": "invoice_sync",
        "status": "success",
        "duration_ms": round((time.perf_counter() - started_at) * 1000),
        "attempt": 1,
        "release": os.environ.get("APP_RELEASE"),
    })
except Exception:
    print({
        "event_id": event_id,
        "error_type": "telemetry_delivery_failed",
    })

The synchronous client uses requests and blocks until the request completes. The published client does not expose a timeout argument, session injection, or automatic retry. For a service with strict latency and connection-pool requirements, call the HTTP API through an application-owned client with explicit timeouts, or isolate the SDK call in a bounded worker.

Use the asynchronous client

import asyncio
import time
import uuid

async def record_job():
    started_at = time.perf_counter()
    await telemetry.log("job_completed", {
        "event_id": str(uuid.uuid4()),
        "job_name": "invoice_sync",
        "status": "success",
        "duration_ms": round((time.perf_counter() - started_at) * 1000),
        "attempt": 1,
    })

asyncio.run(record_job())

TelemetryAsync.log opens an aiohttp session for the request and closes it afterward. It avoids blocking the event loop, but the current package does not expose a shared session, background queue, retry policy, or flush method.

Send a compatible batch

Both clients accept a list of dictionaries:

await telemetry.log("api_request_completed", [
    {
        "event_id": "evt_201",
        "route_template": "/api/projects/:id",
        "status": "success",
        "status_code": 200,
        "latency_ms": 84,
    },
    {
        "event_id": "evt_202",
        "route_template": "/api/projects/:id",
        "status": "failed",
        "status_code": 503,
        "latency_ms": 904,
        "error_type": "dependency_unavailable",
    },
])

Keep every item compatible with the same table schema. Bound any application-owned batch queue and document its overflow and shutdown policy.

Run SQL

Synchronous:

result = telemetry.query("""
    SELECT
      status,
      COUNT(*) AS jobs
    FROM job_completed
    WHERE timestamp_utc >= now() - INTERVAL '24 hours'
    GROUP BY status
    ORDER BY jobs DESC
""")

for row in result.get("data", []):
    print(row["status"], row["jobs"])

Asynchronous:

async def load_summary():
    return await telemetry.query("""
        SELECT status, COUNT(*) AS jobs
        FROM job_completed
        GROUP BY status
        ORDER BY jobs DESC
    """)

These methods use the interactive query endpoint. Use the async Query API directly for long-running JSON or Parquet exports.

Retry and failure policy

Do not retry an unchanged 400 request. For transient connection failures, 429, 502, 503, and 504, use a bounded application retry with exponential backoff and jitter. Reuse the same event_id for the logical event.

For most application analytics, a telemetry outage should not replace a successful customer response. For billing or approved audit workflows, use a durable application-owned outbox. Review event delivery and idempotency.

Verify and troubleshoot

Send known success and failure fixtures, then query:

SELECT timestamp_utc, event_id, job_name, status, duration_ms, error_type
FROM job_completed
ORDER BY timestamp_utc DESC
LIMIT 20;

Confirm types, timestamps, and sensitive-data boundaries. Common failures:

API key is not initialized: call init with a non-empty server-side key.
401 or 403: replace the key or correct its scope.
JSON-decoding exception: inspect the HTTP status and safe response context outside the SDK.
Slow synchronous request: move to the async client or an application-owned HTTP client with a timeout.
Process shutdown: await tracked calls or persist required events; neither client maintains a flushable queue.

Continue with the FastAPI integration, Django and Celery integration, and ingestion troubleshooting.

Use this doc with your coding agent