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.

These examples send fictional values to check SDK delivery. Keep them in telemetry_quickstart; they do not instrument a running application. For real use, record observed values after your chosen operation completes in its own workflow table. Measure duration around the actual operation, not around the telemetry call.

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 an ingest-only key for event producers or a read-scoped key for query-only jobs.

Send an event synchronously

import uuid

event_id = str(uuid.uuid4())

try:
    response = telemetry.log("telemetry_quickstart", {
        "event_id": event_id,
        "job_name": "invoice_sync",
        "status": "success",
        "duration_ms": 912,
        "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 uuid

async def record_job():
    await telemetry.log("telemetry_quickstart", {
        "event_id": str(uuid.uuid4()),
        "job_name": "invoice_sync",
        "status": "success",
        "duration_ms": 912,
        "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:

import asyncio

async def send_batch():
    await telemetry.log("telemetry_quickstart", [
        {
            "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",
        },
    ])

asyncio.run(send_batch())

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 sample_events
    FROM telemetry_quickstart
    WHERE timestamp_utc >= now() - INTERVAL '24 hours'
    GROUP BY status
    ORDER BY sample_events DESC
""")

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

Asynchronous:

async def load_summary():
    return await telemetry.query("""
        SELECT status, COUNT(*) AS sample_events
        FROM telemetry_quickstart
        GROUP BY status
        ORDER BY sample_events 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
FROM telemetry_quickstart
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.