SDK de Python

El paquete telemetry-sh proporciona Telemetry para aplicaciones síncronas y TelemetryAsync para servicios asyncio. Ambos clientes envían cada llamada a método inmediatamente al Telemetry HTTP API. Mantenga la clave API en la configuración del lado del servidor.

Estos ejemplos envían valores ficticios para comprobar la entrega del SDK. Manténlos en telemetry_quickstart; no instrumentan una aplicación en ejecución. Para un uso real, registra los valores observados después de que termine la operación elegida, en su propia tabla de trabajo. Mide la duración de la operación real, no de la llamada a Telemetry.

Instalar e inicializar

python -m pip install telemetry-sh

Código sincrónico:

import os
from telemetry_sh import Telemetry

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

Código asincrónico:

import os
from telemetry_sh import TelemetryAsync

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

init es un método normal para ambos clientes; no lo await. Inicialice una vez por proceso con una clave de solo ingesta para productores de eventos o una clave de alcance de lectura para trabajos de solo consulta.

Enviar un evento sincrónicamente

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

El cliente síncrono utiliza requests y se bloquea hasta que se completa la solicitud. El cliente publicado no expone un argumento de tiempo de espera, inyección de sesión ni reintento automático. Para un servicio con estrictos requisitos de latencia y grupo de conexiones, llame a HTTP API a través de un cliente propiedad de la aplicación con tiempos de espera explícitos o aísle la llamada SDK en un trabajador limitado.

Utilice el cliente asincrónico

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 abre una sesión aiohttp para la solicitud y luego la cierra. Evita bloquear el bucle de eventos, pero el paquete actual no expone una sesión compartida, una cola en segundo plano, una política de reintento o un método de vaciado.

Enviar un lote compatible

Ambos clientes aceptan una lista de diccionarios:

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())

Mantenga todos los elementos compatibles con el mismo esquema de tabla. Vincule cualquier cola por lotes propiedad de la aplicación y documente su política de desbordamiento y cierre.

Ejecute SQL

Sincrónico:

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

Asincrónico:

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

Estos métodos utilizan el punto final de consulta interactiva. Utilice el Consulta asíncrona API directamente para exportaciones JSON o Parquet de larga duración.

Política de reintentos y fallos

No vuelva a intentar una solicitud 400 sin cambios. Para errores de conexión transitorios, 429, 502, 503 y 504, utilice un reintento de aplicación limitado con retroceso y fluctuación exponenciales. Reutilice el mismo event_id para el evento lógico.

Para la mayoría de los análisis de aplicaciones, una interrupción de la telemetría no debería reemplazar una respuesta exitosa del cliente. Para flujos de trabajo de facturación o auditoría aprobados, utilice una bandeja de salida duradera propiedad de la aplicación. Revisar entrega de eventos e idempotencia.

Verificar y solucionar problemas

Envíe accesorios conocidos de éxito y fracaso, luego consulte:

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

Confirme tipos, marcas de tiempo y límites de datos confidenciales. Fallos comunes:

API key is not initialized: llame a init con una clave del lado del servidor que no esté vacía.
401 o 403: sustituir la llave o corregir su alcance.
Excepción de decodificación de JSON: inspeccione el estado de HTTP y el contexto de respuesta segura fuera de SDK.
Solicitud sincrónica lenta: pase al cliente asíncrono o a un cliente HTTP propiedad de la aplicación con un tiempo de espera.
Cierre del proceso: espere llamadas rastreadas o persista en los eventos requeridos; ninguno de los clientes mantiene una cola descartable.

Continúe con Integración rápida de API, Integración de Django y Apio y solución de problemas de ingestión.

Usa esta documentación con tu agente de programación