Python SDK
telemetry-sh 软件包提供用于同步应用的 Telemetry 和用于 asyncio 服务的 TelemetryAsync。两个客户端立即将每个方法调用发送到 Telemetry HTTP API。将 API 密钥保留在服务器端配置中。
安装并初始化
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 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",
})
同步客户端使用 requests 并阻塞，直到请求完成。已发布的客户端不会公开超时参数、会话注入或自动重试。对于具有严格延迟和连接池要求的服务，请通过具有显式超时的应用程序拥有的客户端调用 HTTP API，或在有界工作线程中隔离 SDK 调用。
使用异步客户端
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 针对该请求打开 aiohttp 会话，然后将其关闭。它避免阻塞事件循环，但当前包不公开共享会话、后台队列、重试策略或刷新方法。
发送兼容批次
两个客户端都接受字典列表：
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",
},
])
保持每个项目与相同的表模式兼容。绑定任何应用程序拥有的批处理队列并记录其溢出和关闭策略。
运行SQL
同步：
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"])
异步：
async def load_summary():
return await telemetry.query("""
SELECT status, COUNT(*) AS jobs
FROM job_completed
GROUP BY status
ORDER BY jobs DESC
""")
这些方法使用交互式查询端点。直接使用 异步查询 API 进行长期运行的 JSON 或 Parquet 导出。
重试和失败策略
不要重试未更改的 400 请求。对于瞬时连接故障，429、502、503 和 504，请使用具有指数退避和抖动的有界应用程序重试。对逻辑事件重复使用相同的 event_id。
对于大多数应用程序分析，遥测中断不应取代成功的客户响应。对于计费或批准的审核工作流程，请使用持久的应用程序拥有的发件箱。查看 事件传递和幂等性。
验证并排除故障
发送已知成功和失败的灯具，然后查询：
SELECT timestamp_utc, event_id, job_name, status, duration_ms, error_type
FROM job_completed
ORDER BY timestamp_utc DESC
LIMIT 20;
确认类型、时间戳和敏感数据边界。常见故障：
API key is not initialized：使用非空服务器端密钥调用init。401或403：更换密钥或更正其范围。- JSON-解码异常：检查HTTP状态和SDK外部的安全响应上下文。
- 慢速同步请求：转移到异步客户端或应用程序拥有的 HTTP 客户端并超时。
- 进程关闭：等待跟踪的调用或保留所需的事件；两个客户端都没有维护可刷新队列。
继续使用 FastAPI集成、Django 和 Celery 集成 和 摄取故障排除。