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 整合 和 攝取故障排除。