Flask and SQLAlchemy Telemetry: 境界から検証された行まで
制御されたアプリケーション境界で Flask and SQLAlchemy Telemetry を使用し、イベント コントラクトを小さく保ち、集約ビューを構築する前に既知の結果を検証します。
- 1
結果を選択してください
フラスコ API の信頼性
- 2
契約を定義する
request_id、エンドポイント、route_template、メソッド、および status_code
- 3
境界を計測する
request.path の代わりに Flask の一致する url_rule を使用して、識別子によってカーディナリティやプライバシーの問題が発生しないようにします。
- 4
証拠を検証する
Exercise a known fixture, then inspect api_request_completed for one correctly typed terminal row.
始める前に
前提条件と境界
- telemetry-sh は各 Flask ワーカー プロセスで 1 回初期化されます
- 安定した Flask エンドポイント分類と単調なリクエスト タイマー
- SQLAlchemy イベント リスナーはカウント、期間、安全な操作メタデータに制限されています
配信設定
サーバー側のインストールと初期化
サーバー側キーを使用して、プロセスごとに 1 回、同期コードの場合は Telemetry を、非同期コードの場合は TelemetryAsync を初期化します。 ブラウザ バンドル、クライアントに表示される環境変数、ソース管理、ログ、および例外メッセージに取り込み資格情報が含まれないようにします。
pipのインストール
python -m pip install telemetry-sh- 1制限されたネットワーク動作を持つ再利用可能なサーバー側配信クライアントを 1 つ準備します。
- 2成功、失敗、再試行、またはタイムアウトの境界に結果イベントを追加します。
- 3アラートを有効にする前に、制御されたフィクスチャを送信し、保存されている行を検査します。
スニペット
1 つの構造化されたイベントから始める
ワークフローが完了、失敗、または再試行される場所にこの図形を追加します。次に、実際のフィールドからダッシュボードを構築します。
Flask and SQLAlchemy Telemetryイベント
from flask import g, request
from time import monotonic
@app.before_request
def start_request_timer():
g.request_started_at = monotonic()
g.db_query_count = 0
@app.after_request
def record_request_outcome(response):
route = request.url_rule.rule if request.url_rule else "unmatched"
telemetry.log("api_request_completed", {
"request_id": request.headers.get("X-Request-ID"),
"endpoint": request.endpoint or "unmatched",
"route_template": route,
"method": request.method,
"status_code": response.status_code,
"status": "failed" if response.status_code >= 500 else "success",
"latency_ms": round((monotonic() - g.request_started_at) * 1000),
"db_query_count": g.db_query_count,
"release": app.config.get("APP_RELEASE"),
})
return responseイベント契約
request_id、エンドポイント、route_template、メソッド、および status_code
ステータス、latency_ms、db_query_count、rollback_count、および error_type
サービス、環境、リリース、および承認されたテナントのコンテキスト
実装のチェックポイント
チェックポイント 1
request.path の代わりに Flask の一致する url_rule を使用して、識別子によってカーディナリティやプライバシーの問題が発生しないようにします。
チェックポイント 2
リクエストごとのカウンターを flask.g に保存し、レビュー済みの SQLAlchemy フックから更新します。 SQL テキスト、バインドされたパラメータ、接続文字列、または ORM オブジェクトを決して送信しないでください。
チェックポイント 3
応答の構築後にティアダウン例外が発生する可能性があるため、両方の質問が重要な場合は、リクエストの結果とトランザクションのロールバックを個別のイベントとしてモデル化します。
検証
イベントが到着したことを証明する
既知の成功例と失敗例を実行した後、これを実行します。最終的なイベント コントラクトがスニペットと異なる場合は、フォールバック テーブル名を置き換えます。
Flask and SQLAlchemy Telemetry 検証クエリ
SELECT *
FROM api_request_completed
ORDER BY timestamp_utc DESC
LIMIT 20;実装の参考資料
新しい運用パスを有効にする前に、イベント契約、データ安全性に関するガイダンス、上流の主要ドキュメントを確認してください。
生産境界
結果イベントを小さく、回復可能なものに保つ
このパターンが提供するのは、
- 上流のワークフローの横にある、制限付きの SQL 対応の結果。
- ダッシュボード、アラート、およびイベント間の相関関係の安定したフィールド。
- 成功、失敗、再試行、タイムアウトの動作を検証するためのフィクスチャ駆動のパス。
このパターンでは提供されません
- OTLP エクスポーター、自動収集パイプライン、または詳細なトレースと診断ログの代替。
- ペイロードにイベント ID が含まれているという理由だけで、1 回だけ配信されます。
- 生のプロバイダー ペイロード、ユーザー コンテンツ、資格情報、または規制されたデータを収集する許可。
イベントスキーマの開始点
このワークフローのイベント コントラクト
クエリまたはスニペットを運用環境に適用する前に、行粒度、出力境界、必要なタイプ、プライバシー クラス、サンプル ペイロード、および検証チェックリストを確認してください。
関連製品の機能
このワークフローを続行します アラート
レビューされた信頼性クエリを、所有するしきい値と応答のワークフローにプロモートします。
関連する SQL レシピ
SQL で次の質問に答えてください
このワークフローの構造化フィールドに対してクエリを実行し、結果の例を検査して、有用な回答をダッシュボードまたはアラートに変換します。
ルート別のAPIエラー率の計算
意味のある 5xx エラー率が最も高いのは、どの API ルートですか?
レシピを開くp50、p95、p99 API レイテンシの計算
テール レイテンシが最も悪いエンドポイントはどれですか?
レシピを開く遅いデータベース クエリをフィンガープリントで検出する
調査するには十分なほど常に遅いデータベース操作はどれですか?
レシピを開くデータベーストランザクションのロールバック率の計算
異常な割合のデータベース トランザクションをロールバックするサービスはどれですか?
レシピを開くデータベースクエリを合計時間の影響でランク付けする
どのデータベース操作が最も累積リクエスト時間を消費しますか?
レシピを開く実装ファミリーごとに参照する
関連する統合パターンを比較する
この統合と組み合わせるテンプレート
さらなる統合
MongoDB 操作 Telemetry
ドキュメントやクエリ値を収集せずに、名前付き MongoDB 操作、レイテンシ、結果数、再試行、トランザクション結果、リリース回帰を追跡します。
オープンガイドMySQL クエリとプール Telemetry
生の SQL やパラメータを保存せずに、正規化された MySQL 操作、プール待機、トランザクション結果、制御されたエラー カテゴリを追跡し、リグレッションをリリースします。
オープンガイドプリズマ ORM データベース Telemetry
生のクエリパラメータを収集せずに、Prisma 操作のフィンガープリント、モデルとメソッドのレイテンシー、失敗、結果数、リリース、データベース依存のワークフローを測定します。
オープンガイド