Python SDK

telemetry-sh 软件包提供用于同步应用的 Telemetry 和用于 asyncio 服务的 TelemetryAsync。两个客户端立即将每个方法调用发送到 Telemetry HTTP API。将 API 密钥保留在服务器端配置中。

这些示例发送虚构值来检查 SDK 的数据传送。请保留在 telemetry_quickstart 表中；它们没有为运行中的应用埋点。实际使用时，在选定操作完成后，将观测值记录到自己的工作流表中。应测量实际操作的耗时，而不是 Telemetry 调用的耗时。

安装并初始化

python -m pip install telemetry-sh

同步代码：

import os
from telemetry_sh import Telemetry

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

异步代码：

import os
from telemetry_sh import TelemetryAsync

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

init 是双方客户的正常方法；不要await吧。每个进程使用事件生成器的仅摄取密钥或仅查询作业的读取范围键初始化一次。

同步发送事件

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

同步客户端使用 requests 并阻塞，直到请求完成。已发布的客户端不会公开超时参数、会话注入或自动重试。对于具有严格延迟和连接池要求的服务，请通过具有显式超时的应用程序拥有的客户端调用 HTTP API，或在有界工作线程中隔离 SDK 调用。

使用异步客户端

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 针对该请求打开 aiohttp 会话，然后将其关闭。它避免阻塞事件循环，但当前包不公开共享会话、后台队列、重试策略或刷新方法。

发送兼容批次

两个客户端都接受字典列表：

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

保持每个项目与相同的表模式兼容。绑定任何应用程序拥有的批处理队列并记录其溢出和关闭策略。

运行SQL

同步：

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

异步：

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

这些方法使用交互式查询端点。直接使用异步查询 API 进行长期运行的 JSON 或 Parquet 导出。

重试和失败策略

不要重试未更改的 400 请求。对于瞬时连接故障，429、502、503 和 504，请使用具有指数退避和抖动的有界应用程序重试。对逻辑事件重复使用相同的 event_id。

对于大多数应用程序分析，遥测中断不应取代成功的客户响应。对于计费或批准的审核工作流程，请使用持久的应用程序拥有的发件箱。查看事件传递和幂等性。

验证并排除故障

发送已知成功和失败的灯具，然后查询：

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

确认类型、时间戳和敏感数据边界。常见故障：

API key is not initialized：使用非空服务器端密钥调用init。
401 或 403：更换密钥或更正其范围。
JSON-解码异常：检查HTTP状态和SDK外部的安全响应上下文。
慢速同步请求：转移到异步客户端或应用程序拥有的 HTTP 客户端并超时。
进程关闭：等待跟踪的调用或保留所需的事件；两个客户端都没有维护可刷新队列。

继续使用 FastAPI集成、Django 和 Celery 集成和摄取故障排除。

让编程智能体使用这篇文档