URLがちゃんと生きているかを一目で確認したい——そんな用途に向けて、Python製のURL死活監視ダッシュボードを実際に作ってみました。使うのはFastAPI・httpx・SQLiteの3つだけで、監視対象の登録から手動チェック、最新ステータスと応答時間の一覧、履歴表示までを1ファイルで完結させます。
ローカルのテスト用エンドポイントを監視して、成功と失敗の履歴が積み上がる様子まで確かめた記録です。
FastAPIの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。
動画の内容をテキストで確認する
オープニング。FastAPI・httpx・SQLiteを使ってURL死活監視ダッシュボードを作るカリキュラムを始めます。概要紹介。
FastAPI・httpx・SQLiteの役割と使い方を学ぶURL死活監視ダッシュボードの要件を整理する 完成コードと実行結果を確認する 最後に完成した画面を実際に操作して確かめる 具体的にやること。
監視対象の名前とURLをフォームから登録する すべてチェックボタンで全対象の死活を確認するHTTPステータスコードが400未満かで死活を判定するtime.perf_counterで応答時間をミリ秒で計測する チェック結果をchecksテーブルへ履歴として保存する 実装環境・必須アプリ。
OS：Windows 11 Pro Python：3.13.3シェル：PowerShell 5.1必須アプリ：コードエディター、Webブラウザ、ターミナル パッケージ：pip、fastapi、"uvicorn[standard]"、httpx、python-multipart PythonのFastAPIとhttpxとは。
FastAPI：FastAPIはPythonでWeb APIや画面を手軽に構築できるWebフレームワークhttpx：httpxはPythonからHTTPリクエストを送って応答を受け取れるHTTP通信ライブラリ FastAPIとhttpxで作るURL監視の要点。
targetsテーブルに監視対象を保存するchecksテーブルにチェック履歴を追記するlatest_checkで対象ごとの最新結果を取り出す URL死活監視ダッシュボードの要件定義。
登録した対象がダッシュボードの一覧に表示される 未チェックの対象がpendingの淡色で表示される /probe/up監視でUPとステータス200が記録される /probe/down監視でDOWNとステータス503が記録される チェックごとにchecksテーブルへ1行追記される 履歴ページに確認日時と応答時間が新しい順で並ぶ INTRO: Monaco EditorでURL死活監視ダッシュボードを実装。
コードを1行ずつ入力し、補完と自動インデントを使いながら実行結果を確認します。LINE 001: モジュール全体の説明開始。このファイル全体がFastAPI・httpx・SQLiteを組み合わせたURL死活監視ダッシュボードであることを示すdocstringの書き出しです。
ファイルの用途を最初に明示する役割を持っています。LINE 003: 主な機能の説明1。監視対象URLの登録と、httpxを使った手動チェックで死活状態・HTTPステータス・応答時間を取得することを説明しています。
このあとの実装で実現する機能の概要を伝えています。LINE 004: 主な機能の説明2。SQLiteへの履歴保存と、最新結果や履歴の画面表示までを1つのファイルで完結させることを説明しています。
アプリ全体の設計方針を読者に伝えています。LINE 005: docstringの終端。モジュールの説明文であるdocstringを閉じている行です。
ここまでの3行がまとめてファイル冒頭のコメントとして扱われます。LINE 006: lifespan管理用の関数を読み込み。アプリ起動時と終了時の処理をまとめて書けるasynccontextmanagerを読み込んでいます。
この後のlifespan関数でデータベース初期化に使われます。LINE 007: 日時取得用クラスを読み込み。現在時刻を扱うdatetimeクラスを読み込んでいます。
登録日時やチェック日時をISO形式の文字列として記録する際に使われます。LINE 008: HTMLエスケープ関数を読み込み。HTML特殊文字を安全な表記に変換するescape関数を読み込んでいます。
画面描画時にユーザー入力をそのままHTMLへ埋め込まないようにするために使われます。LINE 009: パス操作用クラスを読み込み。ファイルパスをオブジェクトとして扱えるPathクラスを読み込んでいます。
データベースファイルの保存先パスを組み立てる際に使われます。LINE 010: SQLite操作モジュールの読み込み。Python標準のSQLiteドライバであるsqlite3モジュールを読み込んでいます。
データベースへの接続やSQL実行を行うために必須のモジュールです。LINE 011: 一時ディレクトリ取得モジュールの読み込み。OSの一時ディレクトリを取得できるtempfileモジュールを読み込んでいます。
書き込み可能な場所にデータベースファイルを配置するために利用します。LINE 012: 時間計測モジュールの読み込み。処理時間を計測できるtimeモジュールを読み込んでいます。
URLチェック時の応答時間をミリ秒単位で測定するために使われます。LINE 014: HTTP通信ライブラリの読み込み。HTTPリクエストを送受信できるhttpxライブラリを読み込んでいます。
監視対象URLへ実際にアクセスして死活確認を行う中心的な役割を担います。LINE 015: FastAPI関連クラスの読み込み。FastAPI本体とフォーム受信用のForm、リクエスト情報を扱うRequestを読み込んでいます。
ルーティングやフォーム入力の受け取りに使われます。LINE 016: レスポンスクラスの読み込み。HTML・JSON・リダイレクトの各レスポンスクラスを読み込んでいます。
画面表示やAPI応答、ページ遷移をそれぞれ適切な形式で返すために使われます。LINE 021: データベースファイルのパス定義。一時ディレクトリの中にurl_monitor.dbというファイル名でデータベースパスを定義しています。
実行環境が読み取り専用でも書き込みできる場所を確保する工夫です。LINE 024: DB接続関数の定義開始。SQLiteへの接続処理をまとめたget_conn関数を定義しています。
以降の各データベース操作関数から共通で呼び出される役割を持っています。LINE 025: データベースへの接続。DB_PATHで指定したファイルに対してsqlite3で接続を開始しています。
この接続オブジェクトを通じてSQLの実行が行われます。LINE 026: 行を辞書風に扱う設定。取得した行を列名でアクセスできるようにrow_factoryへsqlite3.Rowを設定しています。
この後の画面描画でchk["is_up"]のような書き方ができるようになります。LINE 027: 接続オブジェクトの返却。作成した接続connを呼び出し元に返しています。
この戻り値を使って他の関数がデータベース操作を行います。LINE 030: DB初期化関数の定義開始。アプリ起動時に呼ばれるinit_db関数を定義しています。
テーブルが存在しない場合に作成する役割を担います。LINE 031: 接続をwith文で取得。get_conn関数で接続を開き、with文を使うことでブロックを抜けるときに自動的に接続が閉じられるようにしています。
安全にリソースを管理するための書き方です。LINE 032: targetsテーブル作成SQLの実行開始。監視対象を保存するtargetsテーブルを作成するSQLの実行を開始しています。
この後の行で具体的な列定義が続きます。LINE 033: targetsテーブルの作成文。テーブルが存在しない場合のみtargetsテーブルを作成するSQL文の先頭部分です。
IF NOT EXISTSにより二重作成を防いでいます。LINE 034: id列の定義。targetsテーブルの主キーとなるid列を、自動採番の整数型として定義しています。
各監視対象を一意に識別するために使われます。LINE 035: name列の定義。監視対象の名前を保存するname列を、必須のテキスト型として定義しています。
画面上で対象を識別する表示名として使われます。LINE 036: url列の定義。監視対象のURLを保存するurl列を、必須のテキスト型として定義しています。
probe_urlでチェックする際の対象アドレスになります。LINE 037: created_at列の定義。登録日時を保存するcreated_at列を、必須のテキスト型として定義しています。
add_target関数から現在日時が書き込まれます。LINE 038: targetsテーブル定義の終了。targetsテーブルの列定義を閉じる括弧です。
ここまでの列定義がまとまって1つのCREATE TABLE文として実行されます。LINE 039: checksテーブル作成SQLの実行開始。チェック履歴を保存するchecksテーブルを作成するSQLの実行を開始しています。
監視結果を蓄積するための重要なテーブルです。LINE 040: checksテーブルの作成文。テーブルが存在しない場合のみchecksテーブルを作成するSQL文の先頭部分です。
監視履歴を記録するための土台になります。LINE 041: id列の定義。checksテーブルの主キーとなるid列を、自動採番の整数型として定義しています。
各チェック結果を一意に識別するために使われます。LINE 042: target_id列の定義。どの監視対象に対するチェックかを示すtarget_id列を、必須の整数型として定義しています。
targetsテーブルとの紐付けに使われます。LINE 043: checked_at列の定義。チェックを実行した日時を保存するchecked_at列を、必須のテキスト型として定義しています。
履歴一覧の表示や並び替えに使われます。LINE 044: is_up列の定義。死活状態を表すis_up列を、必須の整数型として定義しています。
1ならUP、0ならDOWNという形で保存されます。LINE 045: http_status列の定義。取得したHTTPステータスコードを保存するhttp_status列を整数型として定義しています。
値が取得できなかった場合はNULLを許容する列です。LINE 046: response_ms列の定義。応答時間をミリ秒単位で保存するresponse_ms列を実数型として定義しています。
probe_url関数で計測した値がここに保存されます。LINE 047: error列の定義。通信エラーが起きた場合の例外名を保存するerror列をテキスト型として定義しています。
正常時はNULLのままになります。LINE 048: 外部キー制約の定義。target_id列がtargetsテーブルのidを参照する外部キー制約を定義しています。
これにより対象とチェック履歴の関連付けが保たれます。LINE 049: checksテーブル定義の終了。checksテーブルの列定義を閉じる括弧です。
ここまでの定義がまとまって1つのCREATE TABLE文として実行されます。LINE 050: テーブル作成の確定。conn.commit()を呼び出して、ここまでのテーブル作成処理をデータベースに確定させています。
commitを忘れると変更が反映されません。RUN 1/6: テーブルの作成を確認する。init_dbを呼ぶと、targetsとchecksの2つのテーブルが作成されることを確認します。
CHECK 1/6: 途中実行に成功。作成済みテーブル: ['checks', 'sqlite_sequence', 'targets'] RETURN 01: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 054: 監視対象登録関数の定義開始。新しい監視対象を登録するadd_target関数を定義しています。名前とURLを受け取ってデータベースへ保存する役割です。
LINE 055: 接続をwith文で取得。get_conn関数で接続を開き、with文でブロック終了時に自動的に接続を閉じるようにしています。登録処理を安全に行うための書き方です。
LINE 056: INSERT文の実行開始。targetsテーブルへ新しい行を追加するINSERT文の実行を開始しています。この後の行で具体的なSQLとパラメータが続きます。
LINE 057: INSERT文のSQL本体。name・url・created_atの3列に値を挿入するSQL文を記述しています。プレースホルダーを使うことでSQLインジェクションを防いでいます。
LINE 058: 挿入する値の指定。引数のnameとurl、そして現在時刻を秒単位のISO形式に変換した値をタプルとして渡しています。この値がプレースホルダーに順番に埋め込まれます。
LINE 059: INSERT実行の終了。conn.execute呼び出しの括弧を閉じています。ここまでの内容がまとまって1回のSQL実行として処理されます。
LINE 060: 登録内容の確定。conn.commit()を呼び出して、追加した監視対象をデータベースに確定させています。これによりダッシュボードの一覧に反映されます。
LINE 063: 監視対象一覧取得関数の定義開始。登録済みの監視対象をすべて取得するlist_targets関数を定義しています。ダッシュボード表示のもとになるデータを提供します。
LINE 064: 接続をwith文で取得。get_conn関数で接続を開き、with文でブロック終了時に自動的に接続を閉じるようにしています。読み取り専用の処理でも同じ書き方を統一しています。
LINE 065: 全対象の取得と返却。targetsテーブルからid順にすべての行を取得し、fetchallで結果一覧として返しています。この結果がrender_dashboardでの一覧表示に使われます。
RUN 2/6: 監視対象の登録と取得を確認する。add_targetで登録した監視対象が、list_targetsで取り出せることを確認します。CHECK 2/6: 途中実行に成功。
登録件数: 1 登録した名前: 動作確認用サイト RETURN 02: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 068: 最新チェック取得関数の定義開始。
指定した監視対象の最新チェック結果を取得するlatest_check関数を定義しています。ダッシュボードに現在の状態を表示するために使われます。LINE 069: 接続をwith文で取得。
get_conn関数で接続を開き、with文でブロック終了時に自動的に接続を閉じるようにしています。読み取り処理を安全に行うための書き方です。LINE 070: 最新チェック取得SQLの実行開始。
checksテーブルから対象の最新チェックを取得するSQLの実行を開始しています。この後の行で具体的な条件が続きます。LINE 071: 最新1件を絞り込む条件。
target_idで対象を絞り込み、id降順に並べて先頭の1件だけを取得する条件を記述しています。これにより一番新しいチェック結果だけが得られます。LINE 072: 検索条件の値指定。
プレースホルダーに渡すtarget_idの値をタプルとして指定しています。特定の監視対象に対応する行だけを検索するために必要です。LINE 073: 1件だけの結果を取得。
fetchoneを呼び出して該当する1件だけの結果を取得しています。該当行がない場合はNoneが返り、未チェック状態として画面に表示されます。LINE 079: 相対URL解決関数の定義開始。
監視対象に登録されたURLがスラッシュ始まりの相対パスだった場合に、完全なURLへ変換するresolve_url関数を定義しています。モックサーバーへのチェックを可能にする役割です。LINE 080: 相対パスかどうかの判定。
引数のurlがスラッシュで始まっているかどうかを判定しています。相対パスであれば次の行でbase_urlと結合する処理に進みます。LINE 081: ベースURLとの結合。
base_urlの末尾のスラッシュを取り除いたうえで相対パスと結合し、完全なURLを組み立てて返しています。これによりリクエスト元のホスト情報を使ったURLが得られます。LINE 082: 元のURLをそのまま返却。
相対パスでない場合は、引数のurlをそのまま返しています。すでに完全なURLが指定されている場合の処理です。RUN 3/6: 相対URLの補完を確認する。
resolve_urlが、スラッシュで始まる相対パスをbase_urlと連結して絶対URLへ整えることを確認します。CHECK 3/6: 途中実行に成功。補完結果: http://127.0.0.1:8000/probe/up そのまま: https://example.com/health RETURN 03: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 085: 死活チェック関数の定義開始。実際にURLへアクセスして死活状態を確認するprobe_url関数を定義しています。
本アプリの中核となる処理で、タイムアウトも指定できるようになっています。LINE 086: 計測開始時刻の記録。time.perf_counterで処理開始時点の時刻を記録しています。
この値をもとにリクエストにかかった応答時間を計算します。LINE 087: 例外処理の開始。通信エラーが発生する可能性のある処理をtryブロックで囲んでいます。
接続失敗時にアプリ全体が停止しないようにするための備えです。LINE 088: HTTPクライアントの作成。タイムアウト時間とリダイレクト追従を設定したhttpxのクライアントをwith文で作成しています。
このクライアントを使って監視対象URLへリクエストを送ります。LINE 089: HTTPリクエストの送信。httpxクライアントで対象URLへGETリクエストを送信し、そのレスポンスをrespに受け取ります。
ここで実際に監視対象への通信が行われます。LINE 090: 応答時間の計算。perf_counterで取得した開始時刻との差分をミリ秒に変換し、応答時間として求めます。
この値は結果としてあとで利用されます。LINE 091: 結果辞書の開始。チェック結果をまとめる辞書の作成を開始します。
以降の各キーに死活状態や応答時間などの情報を格納していきます。LINE 092: 死活状態の判定。HTTPステータスコードが400未満であればアップとみなし、is_upキーにTrueまたはFalseを設定します。
これが死活監視の中心的な判定です。LINE 093: HTTPステータスの記録。レスポンスから得られたステータスコードをhttp_statusキーに記録します。
後で画面表示や履歴保存に使われます。LINE 094: 応答時間の格納。先ほど計算した応答時間を小数点1桁に丸めてresponse_msキーに格納します。
表示を見やすくするための処理です。LINE 095: エラー情報の初期化。通信が成功した場合はエラーが発生していないため、errorキーにNoneを設定します。
LINE 096: 結果辞書の終了。成功時の結果辞書の定義を終え、この辞書がprobe_urlの戻り値として呼び出し元に返されます。LINE 097: 通信エラーの捕捉。
httpxの通信に関する例外が発生した場合にこのexceptブロックで捕捉し、exc変数に例外オブジェクトを受け取ります。LINE 098: エラー時の応答時間計算。例外が発生した場合でも開始時刻からの経過時間を計算し、エラー時の応答時間として記録します。
LINE 099: エラー結果辞書の開始。通信失敗時に返す結果辞書の作成を開始します。成功時とは異なる内容を各キーに設定していきます。
LINE 100: ダウン状態の設定。通信が失敗したためis_upキーにFalseを設定し、対象がダウンしていることを示します。LINE 101: ステータス不明の設定。
エラーによりHTTPステータスを取得できなかったため、http_statusキーにNoneを設定します。LINE 102: エラー時の応答時間格納。計算した経過時間を小数点1桁に丸めてresponse_msキーに格納し、失敗時でも計測値を保持します。
LINE 103: エラー種別の記録。発生した例外のクラス名をerrorキーに記録し、どの種類のエラーで失敗したかを分かるようにします。LINE 104: エラー結果辞書の終了。
失敗時の結果辞書の定義を終え、この辞書がprobe_urlの戻り値として呼び出し元に返されます。LINE 111: save_check関数の定義。監視対象のIDとチェック結果を受け取り、その内容をチェック履歴としてデータベースへ保存する関数を定義します。
LINE 112: データベース接続の取得。get_conn関数でデータベース接続を取得し、with文によって処理後に確実にクローズされるようにします。LINE 113: SQL実行の開始。
conn.executeを呼び出し、これから実行するINSERT文とパラメータを渡す準備をします。LINE 114: INSERT文の対象テーブル指定。checksテーブルに対してデータを挿入することを示すSQL文の先頭部分です。
LINE 115: 挿入カラムの指定。checksテーブルに挿入する各カラム名を列挙し、どの項目に値を入れるかを明示します。LINE 116: プレースホルダの指定。
実際の値をあとから安全に埋め込むためのプレースホルダ記号をVALUES句に並べます。LINE 117: パラメータタプルの開始。プレースホルダに対応する実際の値をまとめたタプルの定義を開始します。
LINE 118: 対象IDの指定。どの監視対象に対するチェック結果なのかを示すtarget_idを最初のパラメータとして渡します。LINE 119: チェック日時の記録。
現在時刻を秒単位のISO形式文字列に変換し、チェックが行われた日時として記録します。LINE 120: 死活状態の変換。結果のis_upがTrueなら1、Falseなら0に変換し、SQLiteの整数カラムに保存できる形にします。
LINE 121: HTTPステータスのパラメータ化。probe_urlが返したHTTPステータスの値をそのままパラメータとして渡します。LINE 122: 応答時間のパラメータ化。
probe_urlが返した応答時間の値をそのままパラメータとして渡し、履歴として保存します。LINE 123: エラー情報のパラメータ化。probe_urlが返したエラー種別の値をそのままパラメータとして渡し、失敗理由を保存できるようにします。
LINE 124: パラメータタプルの終了。INSERT文に渡すパラメータタプルの定義を終え、SQL実行に必要な値が揃います。LINE 125: execute呼び出しの終了。
conn.executeの呼び出しを閉じ、SQL文とパラメータを合わせてデータベースに実行を依頼します。LINE 126: コミットによる確定。conn.commitを呼び出して、直前のINSERT処理をデータベースに確定して反映させます。
LINE 129: list_history関数の定義。履歴一覧を取得するための関数を定義し、取得件数の上限をlimit引数で受け取れるようにします。LINE 130: データベース接続の取得。
get_conn関数でデータベース接続を取得し、with文で処理後に自動的にクローズされるようにします。LINE 131: クエリ実行の開始。conn.executeを呼び出し、履歴を取得するためのSELECT文の実行を開始します。
LINE 132: 取得カラムの指定。checksテーブルとtargetsテーブルから取得したい各カラムをSELECT句に列挙します。LINE 133: 取得元テーブルの指定。
checksテーブルをcという別名で指定し、以降このエイリアスを使ってカラムを参照します。LINE 134: テーブル結合の指定。targetsテーブルとchecksテーブルをtarget_idとidで結合し、対象名やURLを一緒に取得できるようにします。
LINE 135: 並び順と件数制限の指定。id列を基準に新しい順へ並べ替え、LIMITで取得件数を制限するよう指定します。LINE 136: 件数パラメータの指定。
LIMITに渡す上限件数をパラメータとして指定し、引数のlimit値がそのまま使われます。LINE 137: 全件取得と関数の戻り値。fetchallで検索結果をすべて取得し、その結果をlist_history関数の戻り値として返します。
RUN 4/6: チェック結果の保存と履歴取得を確認する。save_checkで保存したチェック結果が、list_historyで履歴として取り出せることを確認します。CHECK 4/6: 途中実行に成功。
履歴件数: 1 先頭の名前: 履歴確認用 RETURN 04: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 143: STYLE変数の定義開始。
ダッシュボード画面に適用するCSSをまとめた文字列変数STYLEの定義を開始します。LINE 144: 全体レイアウトの指定。body要素にフォントや余白、文字色を指定し、ページ全体の基本的な見た目を整えます。
LINE 145: 見出しと表の基本スタイル。見出しの文字サイズとテーブルの枠線結合や幅、上部の余白を指定し、表を見やすく整えます。LINE 146: セルの罫線と余白の指定。
th要素とtd要素に枠線とパディング、文字揃え、フォントサイズを指定し、表の各セルを整えます。LINE 147: 見出しと状態色の指定。見出しセルの背景色と、UP状態を表す文字の色や太さを指定します。
LINE 148: ダウンと未チェックの色指定。DOWN状態と未チェック状態を表す文字の色を指定し、状態が視覚的に区別できるようにします。LINE 149: フォームと入力欄の指定。
form要素の表示方法とinput要素の余白を指定し、登録フォームの見た目を整えます。LINE 150: ボタンとURL表示の指定。button要素のパディングとカーソル形状、url用クラスの等幅フォントとサイズを指定します。
LINE 151: リンク色とスタイル終了。リンクの文字色を指定し、style要素を閉じてCSS定義全体を完結させます。LINE 152: STYLE変数の定義終了。
文字列の連結を終えて、これまでの各行を組み合わせたSTYLE変数が完成します。LINE 155: render_status関数の定義。チェック結果を受け取り、画面表示用のステータス文字とCSSクラス名の組を返す関数を定義します。
LINE 156: 未チェック判定。checkがNoneであれば、まだ一度もチェックが行われていない状態と判定します。LINE 157: 未チェック時の戻り値。
未チェックの場合には表示文字列「未チェック」とpendingクラス名の組を返します。LINE 158: UP・DOWNの判定と戻り値。is_upの値に応じてUPとupの組、またはDOWNとdownの組を返し、状態表示を切り分けます。
RUN 5/6: ステータス表示ラベルを確認する。render_statusが、死活の状態に応じてUP・DOWN・未チェックの表示ラベルを返すことを確認します。CHECK 5/6: 途中実行に成功。
UP表示: UP up DOWN表示: DOWN down 未チェック表示: 未チェックpending RETURN 05: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 161: render_dashboard関数の定義。
トップ画面のHTML全体を組み立てて返す関数を定義します。LINE 162: 行文字列の初期化。表の各行を組み立てていくための空文字列rowsを用意します。
LINE 163: 監視対象のループ開始。list_targetsで取得した監視対象を1件ずつ取り出し、以降の処理で各行のHTMLを作成していきます。LINE 164: 最新チェックの取得。
対象ごとにlatest_check関数を呼び出し、その対象の最新のチェック結果をchkに取得します。LINE 165: ステータス表示情報の取得。render_statusにchkを渡し、表示用のステータス文字とCSSクラス名をstatusとcssに受け取ります。
LINE 166: HTTPステータス表示の整形。チェックがない場合や値がない場合はハイフンを表示し、それ以外はステータスコードを文字列にして表示用に整えます。LINE 167: 応答時間表示の整形。
チェックがない場合や値がない場合はハイフンを表示し、それ以外は応答時間を小数点1桁の文字列に整えます。LINE 168: 最終確認日時の整形。チェックがない場合はハイフンを表示し、それ以外は最新チェックの日時を表示用の値として取得します。
LINE 169: 表の行への追記開始。対象の名前やURL、ステータスなどを埋め込んだテーブル行のHTML文字列をrowsへ追記していく処理を開始します。LINE 170: 名前セルの生成。
監視対象の名前をエスケープしてtd要素に埋め込み、テーブル行の先頭セルとして出力します。escapeを使うことでHTMLタグとして誤解釈されるのを防いでいます。LINE 171: URLセルの生成。
監視対象のURLをエスケープし、urlクラス付きのtd要素として出力します。urlクラスによって等幅フォントで見やすく表示されます。LINE 172: ステータスセルの生成。
render_statusで決まったCSSクラスとステータス文字列を使って、UP・DOWN・未チェックの状態を色分け表示するセルを作ります。LINE 173: HTTP・応答時間・日時セルの生成。HTTPステータス、応答時間、最終確認日時をそれぞれtd要素にまとめ、行の残りのセルを1行にして出力します。
LINE 174: 1行分のHTMLをrowsへ追加。組み立てたtr要素をrows文字列に追記し、監視対象ごとに1行ずつテーブル行を蓄積していきます。LINE 175: 監視対象なしの判定。
rowsが空文字のままかどうかを調べ、監視対象が1件も登録されていない場合の分岐に入ります。LINE 176: 未登録時のメッセージ行。監視対象がまだ登録されていない場合に表示する案内メッセージの行をrowsへ設定します。
colspanで6列分を1セルにまとめています。LINE 177: ダッシュボードHTMLの返却開始。ここから完成したダッシュボード画面全体のHTML文字列を組み立てて返す処理が始まります。
LINE 178: HTML文書の宣言部。doctype宣言と日本語ページのhtmlタグ、文字コードをUTF-8に指定するmetaタグを出力します。LINE 179: タイトルとスタイルの挿入。
ブラウザタブに表示されるタイトルを設定し、共通スタイル定義STYLEを埋め込んでhead要素を閉じています。LINE 180: 見出しの表示。画面上部に表示するダッシュボードの見出しh1要素を出力します。
LINE 181: 登録フォームの開始タグ。監視対象を新規登録するためのフォームを開始し、POST送信先を/targetsに指定します。LINE 182: 名前入力欄。
監視対象の名前を入力するテキストボックスを配置し、必須入力にすることで空欄送信を防いでいます。LINE 183: URL入力欄の開始。監視対象のURLを入力するinput要素を開始し、name属性をurlに設定しています。
LINE 184: URL入力欄のプレースホルダーと必須指定。入力例をプレースホルダーとして表示し、required属性で入力必須にしてinput要素を完成させます。LINE 185: 登録ボタンとフォーム終了。
登録ボタンを配置し、クリックすると入力内容がフォームとして送信されるようにフォームを閉じています。LINE 186: 一括チェックフォームの開始タグ。全監視対象を一括でチェックするためのフォームを開始し、送信先を/checkに指定します。
LINE 187: チェック実行ボタン。すべての監視対象を一度にチェックするボタンを配置し、フォームを閉じています。LINE 188: テーブルの開始とヘッダー行の開始。
監視対象一覧を表示するテーブルを開始し、ヘッダー行のtrとthead要素を開始します。LINE 189: ヘッダー列の一部。名前・URL・最新ステータス・HTTPステータスの4つの見出しセルを出力します。
LINE 190: ヘッダー列の残りとthead終了。応答時間と最終確認日時の見出しセルを出力し、theadとヘッダー行を閉じます。LINE 191: テーブル本体の挿入。
これまでに組み立てたrows文字列をtbodyに埋め込み、テーブル全体を閉じています。LINE 192: 履歴ページへのリンク。監視履歴ページ/historyへ遷移するリンクを段落として配置します。
LINE 193: HTML文書の終了。body要素とhtml要素を閉じ、ダッシュボード画面のHTML文字列を完成させます。LINE 194: return文の閉じ括弧。
複数行に分けて連結してきた文字列全体をひとつの戻り値としてまとめる括弧を閉じています。RUN 6/6: ダッシュボードのHTML生成を確認する。render_dashboardが、見出しと登録済みの対象を含むダッシュボードのHTMLを生成することを確認します。
CHECK 6/6: 途中実行に成功。見出しを含む: True 登録名を含む: True 未チェックを含む: True RETURN 06: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 197: 履歴描画関数の定義。監視履歴一覧を表示するHTMLを組み立てるrender_history関数を定義します。LINE 198: 行文字列の初期化。
履歴テーブルの各行を積み上げていくための空文字列rowsを用意します。LINE 199: 履歴データの繰り返し取得。list_historyで取得した過去のチェック結果を新しい順に1件ずつ取り出し、ループ処理します。
LINE 200: ステータス文字列の判定。is_upの値に応じてUPまたはDOWNの文字列を決定し、statusに格納します。LINE 201: CSSクラスの判定。
is_upの値に応じてupまたはdownのCSSクラス名を決定し、色分け表示に使います。LINE 202: HTTPステータスの整形。http_statusがNoneの場合はハイフンを、それ以外は文字列化した値を表示用に用意します。
LINE 203: 応答時間の整形。response_msがNoneの場合はハイフンを、それ以外は小数点1桁にフォーマットした値を用意します。LINE 204: 履歴行の組み立て開始。
1件分の履歴データからテーブル行tr要素の文字列を組み立て始めます。LINE 205: 確認日時と名前のセル。チェック日時と監視対象名をエスケープしてそれぞれのセルとして出力します。
LINE 206: URLセルの出力。監視対象のURLをエスケープし、urlクラス付きのセルとして表示します。LINE 207: ステータスセルの出力。
判定済みのCSSクラスとステータス文字列を使い、色分けされたステータスセルを出力します。LINE 208: HTTPと応答時間のセル・行の終了。HTTPステータスと応答時間のセルを出力し、tr要素を閉じて1行分のHTMLを完成させます。
LINE 209: 1行分のHTMLをrowsへ追加。組み立てたtr要素をrowsに追記し、履歴を1行ずつ積み上げていきます。LINE 210: 履歴なしの判定。
rowsが空のままかどうかを確認し、履歴が1件もない場合の分岐に入ります。LINE 211: 履歴なし時のメッセージ行。履歴がまだ登録されていない場合に表示する案内メッセージ行をrowsへ設定します。
LINE 212: 履歴HTMLの返却開始。ここから履歴画面全体のHTML文字列を組み立てて返す処理が始まります。LINE 213: HTML文書の宣言部。
doctype宣言と日本語ページのhtmlタグ、文字コードをUTF-8に指定するmetaタグを出力します。LINE 214: タイトルとスタイルの挿入。履歴ページのタイトルを設定し、共通スタイルSTYLEを埋め込んでhead要素を閉じています。
LINE 215: 見出しの表示。画面上部に表示する監視履歴の見出しh1要素を出力します。LINE 216: テーブルの開始とヘッダー行の開始。
履歴一覧を表示するテーブルを開始し、ヘッダー行のtrとthead要素を開始します。LINE 217: ヘッダー列の一部。確認日時・名前・URL・ステータスの4つの見出しセルを出力します。
LINE 218: ヘッダー列の残りとthead終了。HTTPと応答時間の見出しセルを出力し、theadとヘッダー行を閉じます。LINE 219: テーブル本体の挿入。
組み立てたrows文字列をtbodyに埋め込み、テーブル全体を閉じています。LINE 220: ダッシュボードへ戻るリンク。トップページ/へ戻るためのリンクを段落として配置します。
LINE 221: HTML文書の終了。body要素とhtml要素を閉じ、履歴画面のHTML文字列を完成させます。LINE 222: return文の閉じ括弧。
複数行に分けて連結してきた文字列全体をひとつの戻り値としてまとめる括弧を閉じています。LINE 226: 非同期コンテキストマネージャの指定。次に定義するlifespan関数を非同期コンテキストマネージャとして扱うためのデコレータです。
LINE 227: アプリのライフスパン関数の定義。FastAPIアプリの起動時と終了時の処理をまとめて記述するlifespan関数を非同期関数として定義します。LINE 228: データベース初期化の実行。
アプリ起動時にinit_dbを呼び出し、targetsテーブルとchecksテーブルが確実に存在する状態にします。LINE 229: 起動処理とアプリ実行の分岐。yieldによって初期化処理と終了処理を分け、この位置でアプリ本体が稼働します。
LINE 232: FastAPIアプリの生成。lifespanを組み込んだFastAPIアプリのインスタンスを作成し、以降のルーティング定義で使うappを用意します。LINE 233: ルート一覧の定義。
トップページと履歴ページのパスをリストとしてまとめ、アプリが提供する主要なルートを表しています。LINE 236: トップページのルート指定。GETメソッドで/にアクセスした際にdashboard関数を呼び出し、HTMLResponseとして返すよう指定します。
LINE 237: ダッシュボード表示関数の定義。トップページ表示用のdashboard関数を定義し、後続でrender_dashboardの結果を返す準備をします。LINE 238: ダッシュボードHTMLの返却。
render_dashboardを呼び出して組み立てたHTML文字列をそのままレスポンスとして返します。LINE 241: 登録エンドポイントのルート指定。POSTメソッドで/targetsにアクセスした際にregister関数を呼び出すよう指定します。
LINE 242: 登録関数とフォームパラメータの定義。フォーム送信されたnameとurlを受け取るregister関数を定義し、必須パラメータとしてForm(...)を指定しています。LINE 243: 監視対象の登録処理。
前後の空白を除去したnameとurlを使ってadd_targetを呼び出し、新しい監視対象をデータベースに登録します。LINE 244: 登録後のリダイレクト。登録処理後にトップページ/へ303リダイレクトし、ブラウザが登録直後の一覧画面を再表示できるようにします。
LINE 247: checkエンドポイントの定義。POSTメソッドで/checkへアクセスされたときに実行される処理を登録しています。すべての監視対象を一括でチェックするためのエンドポイントです。
LINE 248: 全対象チェック関数の宣言。check_all関数を定義し、引数としてRequestオブジェクトを受け取ります。このリクエスト情報をもとに、アクセス元のベースURLを取得できるようにしています。
LINE 251: ベースURLの取得。リクエストからアプリ自身のベースURLを文字列として取得しています。この後、相対パスで登録されたURLを絶対URLに変換する際に利用します。
LINE 252: 登録済み対象のループ。list_targets関数で取得したすべての監視対象を1件ずつ順番に処理していきます。登録されている対象の数だけチェックが繰り返されます。
LINE 253: URLの解決とチェックの実行。resolve_urlで対象のURLを絶対URLに変換し、その結果をprobe_urlに渡して死活チェックを実行しています。取得した結果はresult変数に格納されます。
LINE 254: チェック結果の保存。直前で取得したチェック結果を、対象のIDとともにsave_check関数へ渡してSQLiteに保存しています。これにより監視履歴が1件追加されます。
LINE 255: ダッシュボードへのリダイレクト。全対象のチェックが終わったら、トップページへリダイレクトさせています。ステータスコード303を指定することで、フォーム送信後の再送信を防いでいます。
LINE 258: historyエンドポイントの定義。GETメソッドで/historyへアクセスされたときに実行される処理を登録しています。レスポンスの形式はHTMLであることも指定しています。
LINE 259: 履歴表示関数の宣言。history関数を定義しており、監視履歴のページを表示するための処理をこの関数内にまとめています。LINE 260: 履歴HTMLの返却。
render_history関数を呼び出して生成したHTML文字列をそのまま返しています。これにより監視履歴の一覧画面がブラウザに表示されます。LINE 264: probe/upエンドポイントの定義。
GETメソッドで/probe/upへアクセスされたときの処理を登録しています。動作確認用に、常に成功するモックのエンドポイントです。LINE 265: 成功モック関数の宣言。
probe_upという非同期関数を定義しています。監視対象として登録することで、正常に応答するURLの動作を確認できます。LINE 266: 成功レスポンスの返却。
ステータスが正常であることを示すJSONを返しています。デフォルトのステータスコード200が使われるため、probe_urlからはUPと判定されます。LINE 269: probe/downエンドポイントの定義。
GETメソッドで/probe/downへアクセスされたときの処理を登録しています。動作確認用に、常に失敗するモックのエンドポイントです。LINE 270: 失敗モック関数の宣言。
probe_downという非同期関数を定義しています。監視対象として登録することで、異常に応答するURLの動作を確認できます。LINE 271: エラーレスポンスの返却。
エラーであることを示すJSONと、ステータスコード503を返しています。この結果、probe_urlからはDOWNと判定されるようになります。PREVIEW: 完成したURL死活監視ダッシュボードを操作。
入力・送信・結果表示を実際のブラウザ画面で確認します。UPLOAD: 実際の入力データでアプリを操作。監視対象を2件登録して『すべてチェック』を押すと、正常サーバーがUP・障害サーバーがDOWNとして最新ステータスと応答時間の一覧に反映されるのを確認できました。
確認1/4: 監視対象を登録して一覧に表示する。登録フォームへ名前とURLを送信し、ダッシュボードのテーブルに監視対象と未チェック表示が並ぶことを確認します。確認2/4: すべてチェックでUP結果を記録する。
サーバーを起動して/probe/upを監視対象に登録し、すべてチェックを実行して最新結果がUPとステータス200になることを確認します。確認3/4: すべてチェックでDOWN結果を記録する。同じ手順で/probe/downを監視対象に登録し、すべてチェックの結果が失敗を示すDOWNとステータス503になることを確認します。
確認4/4: 監視履歴を新しい順に確認する。/probe/upを2回チェックしたあと履歴ページを開き、監視結果が新しい順に積み上がっていることを確認します。学習内容のまとめ。
@app.getで定義するトップと履歴の画面timeout=5.0で応答待ちの上限を決めるtargetsテーブルに監視対象を保存する データベースファイルを一時ディレクトリ配下に作成する 小さく実行確認しながら完成状態まで段階的に組み立てる エンディング。Python研修はCodeCampでご確認ください。
PythonのFastAPIとhttpxとは
今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。
画面とルーティングを支えるFastAPIの基本
FastAPIはPythonでWeb APIや画面を手軽に構築できるWebフレームワークです。関数にデコレーターを付けるだけでURLと処理が結び付き、少ないコードでルーティングを組み立てられます。今回のダッシュボードでは画面表示やフォーム送信の受け取り、リダイレクトをFastAPIが引き受けます。
起動時にはlifespanという仕組みでデータベースの初期化を1度だけ行う点も特徴です。
ダッシュボードでFastAPIが担当する処理を次の観点で示します
- @app.getで定義するトップと履歴の画面
- @app.postで受ける登録とチェックの処理
- Form(...)で必須にする名前とURLの入力
- RedirectResponseのstatus_code=303指定
- HTMLResponseで返すダッシュボードのHTML
URLへ接続して死活を測るhttpxの働き
httpxはPythonからHTTPリクエストを送り、応答を受け取れるHTTP通信ライブラリです。今回はhttpx.Clientでタイムアウトやリダイレクト追跡を設定し、監視対象のURLへGETリクエストを送ります。返ってきたステータスコードで死活を判定し、送信から受信までの時間をミリ秒に換算して記録する流れです。
接続できないときは例外を捕まえ、失敗の記録として扱う設計になっています。
httpxで死活チェックを行う際の設定と判定を次の観点で示します
- timeout=5.0で応答待ちの上限を決める
- follow_redirects=Trueでリダイレクトを追う
- status_codeが400未満ならUPと判定する
- time.perf_counterで応答時間を計測する
- httpx.HTTPError発生時はDOWNとして扱う
Python・FastAPIで開発する場合の環境構築
この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト（cmd）の両方で使えます。
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install --upgrade pip
.\.venv\Scripts\python.exe -m pip install fastapi "uvicorn[standard]" httpx python-multipart
.\.venv\Scripts\python.exe -c "import sqlite3; print('SQLite', sqlite3.sqlite_version, 'を標準ライブラリから利用可能です')"macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。
python3 -m venv .venv
./.venv/bin/python -m pip install --upgrade pip
./.venv/bin/python -m pip install fastapi "uvicorn[standard]" httpx python-multipart
./.venv/bin/python -c "import sqlite3; print('SQLite', sqlite3.sqlite_version, 'を標準ライブラリから利用可能です')"- SQLiteはPython標準ライブラリのsqlite3モジュールとして同梱されているため個別のpipインストールは不要ですが、上記コマンドでsqlite3モジュールとSQLiteのバージョンを事前に確認し、確実に利用できる状態にしてから起動します。
- DBファイルurl_monitor.dbはアプリ起動時に自動生成されます。
- フォーム送信の受け取りにpython-multipartが必要です。未導入だとForm解析でエラーになります。
- 起動は .\.venv\Scripts\python.exe -m uvicorn url_monitor:app --reloadで行い、ブラウザでhttp://127.0.0.1:8000を開きます。
URL死活監視ダッシュボードの要件定義
目的は、FastAPIとhttpxで監視対象URLを登録し手動チェックの死活結果とHTTPステータスと応答時間を一覧と履歴で確認できるダッシュボードを作ることです。
対象者として、Pythonの基本文法を理解しWebアプリでHTTP監視やSQLite保存の実装を学びたい人を想定しています。
完成物は、FastAPI・httpx・SQLiteで作り登録した複数URLの死活とHTTPステータスと応答時間を一覧と履歴で確認できるURL死活監視ダッシュボードです。
実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。
機能要件
- 監視対象の名前とURLをフォームから登録する
- すべてチェックボタンで全対象の死活を確認する
- HTTPステータスコードが400未満かで死活を判定する
- time.perf_counterで応答時間をミリ秒で計測する
- チェック結果をchecksテーブルへ履歴として保存する
- 各対象の最新チェック結果をトップ画面に一覧表示する
- 全チェック履歴を新しい順に履歴ページへ表示する
- 相対パスのURLをbase_urlで絶対URLへ補完する
- 成功と失敗を返すモックエンドポイントを備える
非機能要件
- データベースファイルを一時ディレクトリ配下に作成する
- targetsとchecksのテーブルを起動時に用意する
- httpxのタイムアウトを5秒に設定する
- follow_redirectsでリダイレクトを追跡する
- HTML出力値をhtml.escapeでエスケープする
- 監視処理全体を1つのPythonファイルにまとめる
- 存在すればテーブルを作り直さず初期化する
- フォーム送信後に303リダイレクトで一覧へ戻す
実装方針
今回はFastAPI・httpx・SQLiteの基本動作を追いやすくするため、URL死活監視ダッシュボード本体を1つのPythonファイルへまとめます。
入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。
URL死活監視ダッシュボードを安全に組み立てるための実装方針は次のとおりです。
- データベースファイルを一時ディレクトリ配下に作成する
- targetsとchecksのテーブルを起動時に用意する
- httpxのタイムアウトを5秒に設定する
- follow_redirectsでリダイレクトを追跡する
- HTML出力値をhtml.escapeでエスケープする
- 監視処理全体を1つのPythonファイルにまとめる
- 存在すればテーブルを作り直さず初期化する
- フォーム送信後に303リダイレクトで一覧へ戻す
完成と判断する条件
- 登録した対象がダッシュボードの一覧に表示される
- 未チェックの対象がpendingの淡色で表示される
- /probe/up監視でUPとステータス200が記録される
- /probe/down監視でDOWNとステータス503が記録される
- チェックごとにchecksテーブルへ1行追記される
- 履歴ページに確認日時と応答時間が新しい順で並ぶ
- 対象が無い場合に監視対象がまだありませんと表示される
URL死活監視ダッシュボードを作る際の重要ポイント
このダッシュボードの中核は、チェック結果をSQLiteへ1件ずつ追記し、最新状態と履歴という2つの見え方で読み出す流れにあります。すべてチェックを押すと、登録済みの各対象に対してprobe_urlが走り、その戻り値をsave_checkがchecksテーブルへ書き込む仕組みです。
トップ画面は対象ごとの最新チェックだけを並べ、履歴ページでは全チェックを新しい順にたどれます。SQLiteは追加のサーバーを立てずにファイル1つで動くため、学習用のダッシュボードと相性の良い保存先です。
チェック結果をSQLiteへ保存し表示するまでの要点を次に挙げます
- targetsテーブルに監視対象を保存する
- checksテーブルにチェック履歴を追記する
- latest_checkで対象ごとの最新結果を取り出す
- list_historyで全履歴を新しい順に取り出す
- JOINで履歴に対象の名前とURLを添える
2つのテーブルで対象と履歴を分ける設計
データはtargetsとchecksの2つのテーブルに分けて持ちます。targetsは監視対象の名前やURLといった変わりにくい情報を1行ずつ保持する場所です。checksはチェックのたびに死活や応答時間を追記していく履歴で、target_idで対象と結び付きます。
役割ごとにテーブルを分けたことで、対象の一覧と履歴の蓄積をそれぞれ素直に書けます。
各テーブルが受け持つ情報は次のとおりです
- targetsは名前とURLと登録日時
- checksは死活とHTTPステータスと応答時間
- checksのtarget_idで対象と結び付ける
最新結果と全履歴を別々に読み出す方法
トップ画面ではlatest_checkがchecksをid降順で1件だけ取り出し、対象ごとの最新状態を表示します。まだ一度もチェックしていない対象はNoneが返るため、未チェックとして淡色で示す作りです。履歴ページのlist_historyはchecksとtargetsをJOINして新しい順に最大100件まで読み出し、名前やURLも一緒に並べます。
画面ごとのチェック結果の読み出し方は次のとおりです
- トップは対象ごとの最新1件を表示する
- 未チェックの対象はpendingで示す
- 履歴は最大100件を新しい順に表示する
相対パスをbase_urlで補って監視する仕組み
チェック実行時はrequest.base_urlでアクセス中のホストを取得し、それをresolve_urlへ渡す流れです。登録されたURLが/で始まる相対パスなら、base_urlと連結して絶対URLに整えたうえでhttpxが取得します。
この工夫のおかげで、同じアプリ内のテスト用エンドポイント/probe/upや/probe/downをそのまま監視対象として試せます。
resolve_urlによるURL補完の分岐は次のとおりです
- /で始まる相対パスはbase_urlと連結する
- スラッシュ以外で始まるURLはそのまま使う
- 同一アプリのモックを監視対象にできる
PythonでURL死活監視ダッシュボードの完成コード
コードは1ファイル完結です。DB初期化、監視対象の登録・取得、httpxによるチェックという役割ごとに関数を分けています。
起動時にinit_dbを呼び、targetsとchecksの2つのテーブルを用意します。データを「対象」と「チェック履歴」に分けて持つのが設計の要点です。
本文ではとくに押さえたい部分だけを抜き出して説明します。全文の逐条解説ではなく、要点を追える形にしました。
このセクションの用語
- row_factory
- sqlite3で取得行の受け取り方を決める設定です。
sqlite3.Rowにすると列名でも値を取り出せます。 - プレースホルダ
- SQL文に値を後から差し込むための
?などの記号です。入力を安全に渡せます。 - perf_counter
-
timeモジュールの高精度なタイマーです。経過時間の計測に向いています。 - follow_redirects
- httpxでリダイレクト先まで自動でたどるかを決める引数です。
- CREATE TABLE
- データを入れる表をデータベースに作成するSQL命令です。
"""FastAPI・httpx・SQLite で作るURL死活監視ダッシュボード。
監視対象URLの登録、httpxによる手動チェック(死活・HTTPステータス・応答時間)、
SQLiteへの履歴保存、最新結果と履歴の表示までを1ファイルで完結させる。
"""
from contextlib import asynccontextmanager
from datetime import datetime
from html import escape
from pathlib import Path
import sqlite3
import tempfile
import time
import httpx
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
# データベースの初期化(SQLite)
# 実行環境によっては本ファイルの配置場所が読み取り専用になっている場合があるため、
# 書き込み可能なテンポラリディレクトリ配下にデータベースファイルを作成する。
DB_PATH = Path(tempfile.gettempdir()) / "url_monitor.db"
def get_conn():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def init_db():
with get_conn() as conn:
conn.execute(
"CREATE TABLE IF NOT EXISTS targets ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" name TEXT NOT NULL,"
" url TEXT NOT NULL,"
" created_at TEXT NOT NULL)"
)
conn.execute(
"CREATE TABLE IF NOT EXISTS checks ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" target_id INTEGER NOT NULL,"
" checked_at TEXT NOT NULL,"
" is_up INTEGER NOT NULL,"
" http_status INTEGER,"
" response_ms REAL,"
" error TEXT,"
" FOREIGN KEY (target_id) REFERENCES targets(id))"
)
conn.commit()
# 監視対象の登録・取得
def add_target(name, url):
with get_conn() as conn:
conn.execute(
"INSERT INTO targets (name, url, created_at) VALUES (?, ?, ?)",
(name, url, datetime.now().isoformat(timespec="seconds")),
)
conn.commit()
def list_targets():
with get_conn() as conn:
return conn.execute("SELECT * FROM targets ORDER BY id").fetchall()
def latest_check(target_id):
with get_conn() as conn:
return conn.execute(
"SELECT * FROM checks WHERE target_id = ? ORDER BY id DESC LIMIT 1",
(target_id,),
).fetchone()
# httpxによる死活チェック
# この probe_url が本アプリの中核で、URLを手動チェックして死活状態とHTTPステータスと応答時間を記録できるようにする役割を担う。
# status_codeが400未満ならUP、接続不可や例外が起きた場合はDOWNとして扱い、time.perf_counterで測った応答時間をミリ秒単位で返す。
def resolve_url(base_url, url):
if url.startswith("/"):
return base_url.rstrip("/") + url
return url
def probe_url(url, timeout=5.0):
started = time.perf_counter()
try:
with httpx.Client(timeout=timeout, follow_redirects=True) as client:
resp = client.get(url)
elapsed = (time.perf_counter() - started) * 1000
return {
"is_up": resp.status_code < 400,
"http_status": resp.status_code,
"response_ms": round(elapsed, 1),
"error": None,
}
except httpx.HTTPError as exc:
elapsed = (time.perf_counter() - started) * 1000
return {
"is_up": False,
"http_status": None,
"response_ms": round(elapsed, 1),
"error": type(exc).__name__,
}
# 履歴の保存と取得(SQLite)
# save_check は probe_url が計測した死活状態・HTTPステータス・応答時間・エラー種別を
# checksテーブルへINSERTすることで、監視結果の履歴をSQLiteへ保存できるようにしている。
# チェックのたびに1行ずつ追記していくため、対象ごとの推移をあとから追跡できる。
def save_check(target_id, result):
with get_conn() as conn:
conn.execute(
"INSERT INTO checks"
" (target_id, checked_at, is_up, http_status, response_ms, error)"
" VALUES (?, ?, ?, ?, ?, ?)",
(
target_id,
datetime.now().isoformat(timespec="seconds"),
1 if result["is_up"] else 0,
result["http_status"],
result["response_ms"],
result["error"],
),
)
conn.commit()
def list_history(limit=100):
with get_conn() as conn:
return conn.execute(
"SELECT c.checked_at, t.name, t.url, c.is_up, c.http_status,"
" c.response_ms, c.error FROM checks c"
" JOIN targets t ON t.id = c.target_id"
" ORDER BY c.id DESC LIMIT ?",
(limit,),
).fetchall()
# HTMLの描画
# render_dashboard と render_history の2つの関数が組み合わさることで、ダッシュボードで最新結果と履歴を表示できるようになっている。
# トップ画面では各対象の最新チェックを一覧にまとめ、/history では全チェックの履歴を新しい順に並べて確認できるようにする。
STYLE = (
"<style>body{font-family:sans-serif;margin:24px;color:#1f2933;}"
"h1{font-size:20px;}table{border-collapse:collapse;width:100%;margin-top:12px;}"
"th,td{border:1px solid #cbd2d9;padding:6px 10px;text-align:left;font-size:14px;}"
"th{background:#f0f4f8;}.up{color:#0b7a35;font-weight:bold;}"
".down{color:#c0143c;font-weight:bold;}.pending{color:#829ab1;}"
"form{display:inline-block;margin-right:8px;}input{padding:6px;margin-right:6px;}"
"button{padding:6px 14px;cursor:pointer;}.url{font-family:monospace;font-size:12px;}"
"a{color:#2563eb;}</style>"
)
def render_status(check):
if check is None:
return "未チェック", "pending"
return ("UP", "up") if check["is_up"] else ("DOWN", "down")
def render_dashboard():
rows = ""
for t in list_targets():
chk = latest_check(t["id"])
status, css = render_status(chk)
http = "-" if not chk or chk["http_status"] is None else str(chk["http_status"])
ms = "-" if not chk or chk["response_ms"] is None else f'{chk["response_ms"]:.1f}'
at = "-" if not chk else chk["checked_at"]
rows += (
f'<tr><td>{escape(t["name"])}</td>'
f'<td class="url">{escape(t["url"])}</td>'
f'<td class="{css}">{status}</td>'
f"<td>{http}</td><td>{ms}</td><td>{escape(at)}</td></tr>"
)
if not rows:
rows = '<tr><td colspan="6">監視対象がまだありません</td></tr>'
return (
"<!doctype html><html lang='ja'><head><meta charset='utf-8'>"
"<title>URL死活監視ダッシュボード</title>" + STYLE + "</head><body>"
"<h1>URL死活監視ダッシュボード</h1>"
"<form id='register-form' action='/targets' method='post'>"
"<input id='target-name' name='name' placeholder='監視対象の名前' required>"
"<input id='target-url' name='url'"
" placeholder='https://example.com/health または /probe/up' required>"
"<button id='register-btn' type='submit'>登録</button></form>"
"<form id='check-form' action='/check' method='post'>"
"<button id='check-all-btn' type='submit'>すべてチェック</button></form>"
"<table id='targets-table'><thead><tr>"
"<th>名前</th><th>URL</th><th>最新ステータス</th><th>HTTPステータス</th>"
"<th>応答時間(ms)</th><th>最終確認日時</th></tr></thead>"
f"<tbody>{rows}</tbody></table>"
"<p><a id='history-link' href='/history'>監視履歴を見る</a></p>"
"</body></html>"
)
def render_history():
rows = ""
for h in list_history():
status = "UP" if h["is_up"] else "DOWN"
css = "up" if h["is_up"] else "down"
http = "-" if h["http_status"] is None else str(h["http_status"])
ms = "-" if h["response_ms"] is None else f'{h["response_ms"]:.1f}'
rows += (
f'<tr><td>{escape(h["checked_at"])}</td><td>{escape(h["name"])}</td>'
f'<td class="url">{escape(h["url"])}</td>'
f'<td class="{css}">{status}</td>'
f"<td>{http}</td><td>{ms}</td></tr>"
)
if not rows:
rows = '<tr><td colspan="6">履歴がまだありません</td></tr>'
return (
"<!doctype html><html lang='ja'><head><meta charset='utf-8'>"
"<title>監視履歴</title>" + STYLE + "</head><body>"
"<h1>監視履歴</h1>"
"<table id='history-table'><thead><tr>"
"<th>確認日時</th><th>名前</th><th>URL</th><th>ステータス</th>"
"<th>HTTP</th><th>応答時間(ms)</th></tr></thead>"
f"<tbody>{rows}</tbody></table>"
"<p><a href='/'>ダッシュボードへ戻る</a></p>"
"</body></html>"
)
# FastAPIアプリとルーティング
@asynccontextmanager
async def lifespan(app):
init_db()
yield
app = FastAPI(lifespan=lifespan)
ROUTES = ["/", "/history"]
@app.get("/", response_class=HTMLResponse)
def dashboard():
return render_dashboard()
@app.post("/targets")
def register(name: str = Form(...), url: str = Form(...)):
add_target(name.strip(), url.strip())
return RedirectResponse("/", status_code=303)
@app.post("/check")
def check_all(request: Request):
# ここで各対象に対して probe_url を呼び出すことで、URLを手動チェックして死活状態とHTTPステータスと応答時間を記録できるようにし、
# その結果を save_check で監視結果の履歴としてSQLiteへ保存できるようにしている。
base = str(request.base_url)
for t in list_targets():
result = probe_url(resolve_url(base, t["url"]))
save_check(t["id"], result)
return RedirectResponse("/", status_code=303)
@app.get("/history", response_class=HTMLResponse)
def history():
return render_history()
# ローカルテスト用のモックサーバー(成功/失敗を再現)
@app.get("/probe/up")
async def probe_up():
return JSONResponse({"status": "ok"})
@app.get("/probe/down")
async def probe_down():
return JSONResponse({"status": "error"}, status_code=503)
コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。
DB_PATHをtempfileで決める
DB_PATH = Path(tempfile.gettempdir()) / "url_monitor.db"DBファイルの置き場所を、書き込みできる一時ディレクトリ配下に決めています。配置場所が読み取り専用でも保存に失敗しないための工夫です。
get_connとrow_factoryで接続する
def get_conn():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return connsqlite3.connectでDBに接続する共通関数です。row_factoryにsqlite3.Rowを指定すると取得結果を列名でも取り出せて、画面表示のコードが読みやすくなります。
init_dbでtargetsテーブルを作る
conn.execute(
"CREATE TABLE IF NOT EXISTS targets ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" name TEXT NOT NULL,"
" url TEXT NOT NULL,"
" created_at TEXT NOT NULL)"
)監視対象を保存するtargetsテーブルを作ります。IF NOT EXISTSを付けているので、すでにあれば作り直さず、起動のたびに実行しても安全です。
add_targetのプレースホルダ登録
conn.execute(
"INSERT INTO targets (name, url, created_at) VALUES (?, ?, ?)",
(name, url, datetime.now().isoformat(timespec="seconds")),
)登録フォームの入力をtargetsへ1行追加します。値を?のプレースホルダで渡すことで、SQLインジェクションを避けつつ安全に保存できます。
latest_checkで最新の1件を取る
return conn.execute(
"SELECT * FROM checks WHERE target_id = ? ORDER BY id DESC LIMIT 1",
(target_id,),
).fetchone()各対象の最新チェックを1件だけ取り出します。ORDER BY id DESC LIMIT 1で新しい順に並べ先頭のみ取得し、ダッシュボードの現在ステータスに使います。
resolve_urlで相対パスを解決する
def resolve_url(base_url, url):
if url.startswith("/"):
return base_url.rstrip("/") + url
return url/probe/upのような相対パスを、ダッシュボードのホスト名と連結して完全なURLにします。ローカルのテスト用エンドポイントを手軽に監視するための処理です。
probe_urlのhttpx.Clientと応答時間
def probe_url(url, timeout=5.0):
started = time.perf_counter()
try:
with httpx.Client(timeout=timeout, follow_redirects=True) as client:
resp = client.get(url)
elapsed = (time.perf_counter() - started) * 1000死活チェックの中核です。httpx.ClientでGETし、time.perf_counterで前後の時間差を測って応答時間をミリ秒に換算します。follow_redirects=Trueでリダイレクト先まで追いかけます。
参考：
©httpx公式ドキュメント (Advanced Usage)If you do anything more than experimentation, one-off scripts, or prototypes, then you should use a Client instance.
URL死活監視ダッシュボードの動作確認
起動後、監視対象を2件登録して動作を確かめました。正常系は/probe/up(200)、障害系は/probe/down(503)という同じアプリ内のエンドポイントを使い、外部通信なしで再現しています。
『すべてチェック』を押すと、正常サーバーがUP・障害サーバーがDOWNとして最新ステータスと応答時間の一覧に反映されるのを確認できました。
/historyを開くと、チェック結果が新しい順に並びます。成功と失敗の両方が履歴に積み上がる様子も確かめられました。
このセクションの用語
- UP/DOWN
- URLが正常に応答する状態がUP、応答できない状態がDOWNです。
- ステータスコード
- HTTPの応答に付く3桁の数字です。200は成功、503は利用不可などを表します。
画面で確認できた挙動の一覧です。
- 操作前のダッシュボードは監視対象が未登録の空の状態から始まります
- 『すべてチェック』の後、UPの行にはHTTPステータス200と応答時間(ms)が並びます
- 障害用の
/probe/downはDOWN判定になり、HTTPステータス503が表示されます -
/historyでは成功と失敗のチェックが新しい順に一覧されます
PythonのFastAPI・httpx・SQLiteのエラー対処
動かし始めのつまずきは、ライブラリの導入とDBファイルの扱い、そして接続失敗まわりに集中しがちです。
監視ツールの性質上、接続失敗は「エラーで落とす」のではなくDOWNとして記録するのが正しい振る舞いです。例外を捕まえて履歴に残す設計にしておきます。
このセクションの用語
- timeout
- 接続や読み取りの待機時間の上限です。処理全体の厳密な上限時間とは限りません。
- 仮想環境
- プロジェクトごとにライブラリを分離できるPythonの実行環境です。
| エラー例 | 原因 | 対処 |
|---|---|---|
| ModuleNotFoundError: No module named 'httpx' | httpxが未インストール | 仮想環境を有効化してhttpxを入れ直す |
| sqlite3のimportに失敗する | Pythonにsqlite3が組み込まれていない |
python -c "import sqlite3"で確認しSQLite付きのPythonを使う |
| httpx.ConnectError | 対象URLへ接続できない | URLとポートを見直し、DOWN扱いとしてerrorに記録する |
| httpx.ConnectTimeout / ReadTimeout | 応答が遅くtimeoutを超過 |
timeoutの値を調整し、例外を握ってDOWN判定にする |
| sqlite3.OperationalError: unable to open database file | DBの保存先が書き込み不可 |
tempfile.gettempdir()配下など書き込み可能な場所へ置く |
URL死活監視ダッシュボードで注意したい点
死活監視で迷いやすいのが、どこからをUPとみなすかです。本アプリはstatus_codeが400未満ならUP、接続不可や例外はDOWNとして扱います。
この「HTTP400未満を成功」という線引きは覚えておくと安心です。参考までにrequestsのResponse.okも、200番台に限らずHTTP400未満でTrueになります。
もうひとつは応答時間の測り方です。time.timeではなくtime.perf_counterを使うと、経過時間をより安定して測れます。
ポイントとしては、UP判定の基準とDBの保存先というつまずきやすい勘所をまとめておきます。
UP判定：HTTP400未満を成功扱い
DOWN記録：例外はDOWNとして保存
DB保存先：書き込み可能な場所へ配置
URL死活監視ダッシュボードの活用例
このダッシュボードは、URLを並べて一括で生死を見たいときに向いています。監視対象をtargetsに足すだけで対象を増やせます。
履歴はchecksに蓄積されるため、いつからDOWNになったかを後から振り返ることが可能です。障害の切り分けや報告の材料にもできます。
このセクションの用語
- 外形監視
- 利用者と同じ外側からURLへアクセスして稼働を確認する監視方法です。
- ヘルスチェック
- サービスが正常に動いているかを確かめるための専用の応答口です。
| 使える場面 | 具体的な使い方 |
|---|---|
| 自作Webサービスの外形監視 | 公開URLを登録し、定期的に『すべてチェック』してUP/DOWNと応答時間を把握する |
| 社内APIの疎通確認 | APIのヘルスチェック用エンドポイントを対象に登録し、障害の一次切り分けに使う |
| リリース直後の動作確認 | デプロイ後に主要URLをまとめてチェックし、履歴で成功と失敗の推移を追う |
| FastAPIの学習用教材 | 登録・チェック・履歴という一連の流れを写経してWebアプリの基礎を学ぶ |
URL死活監視ダッシュボード開発のまとめ
FastAPI・httpx・SQLiteだけで、登録から手動チェック、履歴表示までを1ファイルにまとめられました。外部サービスに頼らない自己完結の構成です。
実際に監視対象を2件登録して『すべてチェック』を押すと、正常サーバーがUP・障害サーバーがDOWNとして最新ステータスと応答時間の一覧に反映されるのを確認できました。
まずはローカルの/probe/upと/probe/downで挙動をつかみ、慣れたら自分のサービスのURLへ広げていくのがおすすめです。
参考にした一次情報
- ^ FastAPI公式ドキュメント. https://fastapi.tiangolo.com/, (参照26-07-30).
- ^ HTTPX公式ドキュメント. https://www.python-httpx.org/, (参照26-07-30).
- ^ Python標準ライブラリsqlite3. https://docs.python.org/3/library/sqlite3.html, (参照26-07-30).
- ^ Uvicorn公式サイト. https://www.uvicorn.org/, (参照26-07-30).
※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。
ITやプログラミングに関するコラム
【Python】FastAPIで料金プラン見積もりシミュレーターを作ってみた
【Python】pandasとmatplotlibで在庫データのABC分析と構成比を可視化してみた
【Python】Flaskで社内FAQをカテゴリ検索できるWebアプリを作ってみた
【Python】argparseでJSON整形・構文検証・キー検索CLIを試してみた
【Python】NumPyとmatplotlibでモンテカルロ法による円周率推定と収束過程の可視化を試してみた
【Python】Playwrightでスクレイピングを試してみた
【CSS】notで複数の件を除外する方法
【Git】remote設定を変更する方法
【VBA】コメントアウトを設定する方法
x86とx64の違いを分かりやすく解説
ITやプログラミングに関するニュース
VercelがAI GatewayにSeedream 5.0 Proを追加、AI SDKのモデル指定で画像生成と編集が可能に
AWSがAmazon LocationのPlaces APIを強化、住所表記の指定と移動手段別の検索が可能に
VercelがトレースにTree・Waterfallビューを追加、ログ画面で処理の階層と所要時間を確認可能に
Googleがエージェント評価の再考を提唱、難易度を情報量で測るDiscovery Benchを解説
Google CloudがCloud Runサンドボックスを公開プレビューで提供、サービスヘルスは一般提供に
Google Cloud EMEAが英国金融の重要第三者に指定、イングランド銀行・PRA・FCAの直接監督下に
AWS DMS Schema ConversionがSQL Serverのオフライン変換に対応、ソースDBへ接続せずスキーマを変換可能に
EC2 G7インスタンスが米国東部（バージニア北部）で利用可能に、G6比でAI推論性能が最大4.6倍
SageMaker HyperPodが継続プロビジョニングでのAMIベース構成に対応、S3のスクリプト管理なしでSlurmクラスターを作成可能に
AWSがEMR on EKSでSparkトラブルシューティングエージェントに対応、失敗ジョブの原因分析を自然言語で依頼可能に
