Python SDK
の telemetry-sh パッケージが提供する Telemetry 同期アプリケーションの場合と TelemetryAsync のために asyncio サービス。両方のクライアントは、各メソッド呼び出しを 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 それ。イベント プロデューサーの場合は書き込みスコープのキー、クエリ専用ジョブの場合は読み取りスコープのキーを使用して、プロセスごとに 1 回初期化します。
イベントを同期的に送信する
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 の統合、そして 取り込みのトラブルシューティング.