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.

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 alcance de escritura para productores de eventos o una clave de alcance de lectura para trabajos de solo consulta.

Enviar un evento sincrónicamente

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

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 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 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:

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

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

Asincrónico:

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