【Python】FastAPIとSQLiteとBackgroundTasksでWebhook受信アプリを作ってみた

【Python】FastAPIとSQLiteとBackgroundTasksでWebhook受信アプリを作ってみた

公開: 更新:
CodeCampが提供するDX人材育成が可能なプログラミングやITが学べる公開講座

外部サービスから届くWebhookを受け取り、あとから中身を見返せる小さなWebアプリをPythonで作りました。使うのはFastAPISQLiteBackgroundTasksの3つだけで、追加のフロントエンドや外部DBが不要な自己完結型です。

POST /webhookでイベントを受信し、応答をすぐ返してから背景処理でSQLiteへ保存し、GET /の履歴画面で受信日時・イベント種別・JSONペイロード・処理状態を一覧します。初心者でも追体験できるよう、コードを断片ごとに噛み砕いて解説します。

FastAPIの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。

【Python】FastAPIとSQLiteとBackgroundTasksでWebhook受信アプリを作ってみた|フルカリキュラム
動画の内容をテキストで確認する

オープニング。FastAPIとSQLiteを使ってWebhook受信アプリを作るカリキュラムを始めます。概要紹介。

FastAPIとSQLiteの役割と使い方を学ぶWebhook受信アプリの要件を整理する 完成コードと実行結果を確認する 最後に完成した画面を実際に操作して確かめる 具体的にやること。

POST /webhookでWebhookのJSONペイロードを受信する 受信直後に202 Acceptedを返すBackgroundTasksで応答後にSQLiteへ1件保存する イベント種別をクエリパラメータかヘッダーから取得する 取得できないときイベント種別をunknownとする 実装環境・必須アプリ。

OS:Windows 11 Pro Python:3.13.3シェル:PowerShell 5.1必須アプリ:コードエディター、Webブラウザ、ターミナル パッケージ:pip、fastapi、"uvicorn[standard]" PythonのFastAPIとSQLiteとは。

FastAPI:FastAPIは、Pythonで軽量なWeb APIやWebアプリを素早く構築できるWebフレームワークSQLite:SQLiteは、Python標準のsqlite3モジュールから扱える軽量な組み込みデータベース FastAPIで作るWebhook受信アプリの要点。

生ボディを読み取りイベント種別と受信日時を決めるBackgroundTasksへ保存関数を登録する202 Acceptedを先に返す Webhook受信アプリの要件定義。

POST /webhookが202 Acceptedを返す 背景処理の後にSQLiteの受信イベントが1件増える 妥当なJSONの受信で処理状態が処理済になる 壊れたJSONの受信で処理状態がエラーになるGET /で受信履歴の表が新しい順に並ぶ 履歴が無いときまだ受信履歴はありませんと表示する INTRO: Monaco EditorでWebhook受信アプリを実装。

コードを1行ずつ入力し、補完と自動インデントを使いながら実行結果を確認します。LINE 001: jsonモジュールの読み込み。JSON文字列を解析したり整形したりするための標準ライブラリjsonを読み込んでいます。

この後の処理でペイロードの妥当性チェックに使われます。LINE 002: sqlite3モジュールの読み込み。軽量なデータベースであるSQLiteを操作するための標準ライブラリを読み込んでいます。

受信履歴の保存や読み出しに使用します。LINE 003: tempfileモジュールの読み込み。OSが用意する一時ディレクトリのパスを取得するための標準ライブラリを読み込んでいます。

データベースファイルの保存場所を決めるために使います。LINE 004: asynccontextmanagerの読み込み。非同期処理向けのコンテキストマネージャを作成するための機能を読み込んでいます。

アプリ起動時の初期化処理を定義する際に利用します。LINE 005: datetimeクラスの読み込み。現在の日時を取得するためのdatetimeクラスを読み込んでいます。

Webhookを受信した時刻を記録する際に使用します。LINE 006: escape関数の読み込み。HTML表示時に特殊文字を安全な文字列へ変換するescape関数を読み込んでいます。

画面にデータを表示する際の安全対策として使います。LINE 007: Pathクラスの読み込み。ファイルパスをオブジェクトとして扱いやすくするPathクラスを読み込んでいます。

データベースファイルのパスを組み立てる際に使用します。LINE 009: FastAPI関連機能の読み込み。Webアプリの本体であるFastAPIクラスや、リクエスト情報を扱うRequest、背景処理を扱うBackgroundTasksを読み込んでいます。

この後のAPI定義で使用します。LINE 010: レスポンス用クラスの読み込み。HTML形式やJSON形式でレスポンスを返すためのクラスを読み込んでいます。

画面表示やAPI応答の作成に使用します。LINE 015: データベースファイルの保存先を決定。OSの一時ディレクトリの中にwebhooks.dbというファイル名でデータベースの保存場所を決めています。

実行環境が変わっても書き込みできる場所を確保しています。RUN 1/6: DBファイルの配置先を確認する。一時ディレクトリ配下に置くDBファイルのパスが決まっていることを確認します。

CHECK 1/6: 途中実行に成功。DBファイルの場所: /tmp/webhooks.db RETURN 01: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。

LINE 018: データベース初期化関数の定義。init_dbという関数を定義し、アプリ起動時にテーブルを準備する処理をまとめています。戻り値がないことを型注釈で示しています。

LINE 019: データベースへの接続。DB_PATHで指定した場所のSQLiteデータベースに接続し、接続オブジェクトをconnという変数に格納しています。この後のSQL実行に使います。

LINE 020: SQL文の実行開始。接続オブジェクトを使ってSQL文を実行しています。テーブルを作成するための命令をこの後の行で渡しています。

LINE 021: SQL文の記述開始。複数行にわたるSQL文の始まりを示しています。テーブル作成のための命令文をこれ以降の行で記述しています。

LINE 022: テーブル作成命令。webhook_eventsという名前のテーブルが存在しない場合にのみ作成する命令を記述しています。既に存在する場合はエラーにならずそのまま利用されます。

LINE 023: 主キー列の定義。idという列を自動採番される主キーとして定義しています。各レコードを一意に識別するために使われます。

LINE 024: 受信日時列の定義。received_atという列を必須の文字列型として定義しています。Webhookを受信した日時を保存するための列です。

LINE 025: イベント種別列の定義。event_typeという列を必須の文字列型として定義しています。受信したイベントの種類を保存するための列です。

LINE 026: ペイロード列の定義。payloadという列を必須の文字列型として定義しています。受信したJSONの内容そのものを保存するための列です。

LINE 027: 処理状態列の定義。statusという列を必須の文字列型として定義しています。JSONが正しく処理できたかどうかの状態を保存するための列です。

LINE 028: テーブル定義の閉じ括弧。CREATE TABLE文で定義した列の並びを閉じています。ここまでの内容でテーブルの構造が確定します。

LINE 029: SQL文の終了。複数行にわたるSQL文の終わりを示しています。ここまでの記述でテーブル作成命令が完成します。

LINE 030: execute呼び出しの閉じ括弧。conn.executeメソッドへ渡すSQL文の指定を閉じています。この行でメソッド呼び出し自体が完了します。

LINE 031: 変更の確定。ここまでのデータベース操作を確定させるためにcommitを呼び出しています。テーブル作成の内容が実際にファイルへ書き込まれます。

LINE 032: データベース接続の終了。使い終えたデータベース接続を閉じています。リソースを解放して安全に処理を終えるための後片付けです。

RUN 2/6: テーブル作成関数を試す。init_dbを実行し、webhook_eventsテーブルが用意されることを確認します。CHECK 2/6: 途中実行に成功。

作成済みテーブル: [('webhook_events',), ('sqlite_sequence',)] RETURN 02: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 035: 保存処理関数の定義。

受信日時、イベント種別、ペイロードを受け取り、SQLiteへ保存するsave_webhook_event_to_sqlite関数を定義しています。この関数は背景処理として呼び出されます。LINE 036: 関数の説明文の開始。

この関数がどのような目的で使われるかを説明するドキュメント文字列の書き出しです。処理内容を後続の行で補足しています。LINE 038: 背景処理の説明。

BackgroundTasksから呼び出されることと、レスポンスを返した後に実行される非同期処理であることを説明しています。LINE 039: 保存処理の内容説明。SQLiteへ1件のデータを挿入する処理であることと、JSONの妥当性によって状態が変わることを説明しています。

LINE 040: 状態分岐の説明。JSONとして正しければ処理済、壊れていればエラーとして保存されることを説明しています。LINE 041: 説明文の終了。

ドキュメント文字列を閉じています。ここまでの説明で関数の役割が明示されています。LINE 043: 例外処理の開始。

ペイロードの解析中にエラーが起きても処理が止まらないようにtry文で囲んでいます。この後の行で実際の解析を行います。LINE 044: JSON文字列の解析。

受け取った文字列をjson.loadsでPythonのデータへ変換しています。変換に成功すれば正しいJSONであると判断できます。LINE 045: JSONの整形と正規化。

解析できたデータをキーの順序を揃えた形で文字列に整形し直しています。保存時のデータ形式を統一するために行っています。LINE 046: 処理済状態の設定。

JSONの解析に成功した場合の処理状態として「処理済」という文字列を設定しています。LINE 047: 例外発生時の捕捉。JSONの解析に失敗した場合に発生する値エラーや型エラーを捕まえています。

この後の行でエラー時の処理を行います。LINE 048: 元の文字列をそのまま保持。解析に失敗した場合は整形せず、受け取った文字列をそのまま保存対象として設定しています。

LINE 049: エラー状態の設定。JSONの解析に失敗した場合の処理状態として「エラー」という文字列を設定しています。LINE 052: データベースへの再接続。

保存処理のためにあらためてSQLiteデータベースへ接続しています。ここから先の行でデータの挿入を行います。LINE 053: 挿入SQLの実行開始。

データベースへ新しいレコードを挿入するためのSQL文を実行しています。この後の行で具体的な命令文とパラメータを指定しています。LINE 054: 挿入対象の列指定。

webhook_eventsテーブルに挿入する列として受信日時、イベント種別、ペイロード、状態の4つを指定しています。LINE 055: プレースホルダーの指定。挿入する値の位置を示すプレースホルダーを記述しています。

実際の値は後続の行で渡されます。LINE 056: 挿入する値の指定。受信日時、イベント種別、整形済みのペイロード、処理状態をタプルとしてまとめ、プレースホルダーへ渡しています。

LINE 057: execute呼び出しの閉じ括弧。挿入命令とパラメータの指定を閉じています。この行でINSERT文の呼び出しが完了します。

LINE 058: 変更の確定。挿入したデータをデータベースファイルへ実際に反映するためにcommitを呼び出しています。LINE 059: データベース接続の終了。

保存処理で使ったデータベース接続を閉じています。処理が終わった後の後片付けとして行っています。RUN 3/6: 保存関数で1件記録する。

save_webhook_event_to_sqliteを呼び、SQLiteへイベントが保存されることを確認します。CHECK 3/6: 途中実行に成功。保存件数: 1 最新の行: ('order.created', '処理済') RETURN 03: エディターへ戻りました。

入力済みのコードを維持したまま、次の実装へ進みます。LINE 063: 非同期コンテキストマネージャの指定。次に定義する関数を非同期のコンテキストマネージャとして扱うためのデコレータです。

アプリの起動と終了のタイミングを制御するために使います。LINE 064: ライフスパン関数の定義。アプリの起動時と終了時に実行される処理をまとめるlifespan関数を定義しています。

引数としてFastAPIアプリ本体を受け取ります。LINE 065: データベースの初期化呼び出し。アプリ起動時にinit_db関数を呼び出し、テーブルが存在しない場合に作成しています。

これによりアプリ利用前にデータベースの準備が整います。LINE 066: 起動処理と終了処理の区切り。yield文を境にして、これより前が起動時の処理、これより後が終了時の処理として扱われます。

ここでは特別な終了処理は行っていません。LINE 069: FastAPIアプリの生成。FastAPIのインスタンスを作成し、appという変数に格納しています。

先ほど定義したlifespan関数を指定し、起動時にテーブル作成が行われるようにしています。LINE 070: ルート一覧の定義。アプリが提供する画面のパスをROUTESという変数にリストとして保持しています。

トップページのパスが登録されています。LINE 074: POSTルートの定義。次に定義する関数をパス/webhookへのPOSTリクエストを処理するエンドポイントとして登録するデコレータです。

LINE 075: Webhook受信関数の定義。receive_webhookという非同期関数を定義し、リクエスト情報と背景処理用のオブジェクトを引数として受け取っています。LINE 076: リクエストボディの取得。

送られてきたリクエストの本文データを非同期に取得し、rawという変数に格納しています。LINE 077: ペイロードの文字列化。取得したバイト列をUTF-8の文字列にデコードし、前後の空白を取り除いています。

中身が空だった場合は空のJSONオブジェクトを代わりに設定しています。LINE 078: イベント種別の決定開始。イベントの種類を判定するための処理を開始しています。

この後の行で複数の取得元から順番に値を探します。LINE 079: クエリパラメータからの取得。URLのクエリパラメータからtypeという値を取得しています。

ここで値が見つかればイベント種別として採用されます。LINE 080: リクエストヘッダーからの取得。クエリパラメータに値がなかった場合に、ヘッダーのx-event-typeという項目からイベント種別を取得しています。

LINE 081: デフォルト値の設定。クエリパラメータにもヘッダーにも値がなかった場合の代わりとして、unknownという文字列を設定しています。LINE 082: イベント種別決定の終了。

複数の候補から値を探す処理のまとまりを閉じています。ここでevent_typeという変数に最終的な値が格納されます。LINE 083: 受信日時の記録。

現在の日時を取得し、決められた書式の文字列に変換してreceived_atという変数に格納しています。データベース保存時の日時として使われます。LINE 086: 背景処理の登録。

受信したデータをすぐには保存せず、レスポンスを返した後に実行されるようbackground_tasksへ保存処理を登録しています。処理する関数と必要な引数を一緒に渡しています。LINE 087: 受付完了レスポンスの返却。

クライアントに対して受け付けが完了したことを示すJSONレスポンスを、ステータスコード202とともに返しています。RUN 4/6: Webhook受信エンドポイントを試す。POST /webhookへ送信し、202 Acceptedが返ることを確認します。

CHECK 4/6: 途中実行に成功。応答コード: 202 応答本文: {'status': 'accepted'} RETURN 04: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。

LINE 091: 履歴HTML生成関数の定義。受信履歴のテーブル行を組み立てて文字列として返す関数を定義しています。この関数はデータベースの内容を画面表示用のHTMLに変換する役割を持ちます。

LINE 092: データベースへの接続。SQLiteのデータベースファイルに接続しています。この接続を使って、これから受信履歴を読み出す処理を行います。

LINE 093: SQL文の実行開始。受信履歴を取得するためのSELECT文を実行しています。取得する列として受信日時・イベント種別・ペイロード・処理状態を指定しています。

LINE 094: 取得する列の指定。SELECT文で取り出す列名を並べています。この4つの列が、後でテーブルの各行として表示される情報になります。

LINE 095: 抽出元と並び順の指定。webhook_eventsテーブルからデータを取得し、id列の降順で並べています。これにより最新の受信履歴が一番上に表示されます。

LINE 096: 結果を全件取得。実行したSQL文の結果をすべて取得し、rowsという変数に格納しています。ここで得られる値は行ごとのタプルのリストです。

LINE 097: データベース接続を閉じる。データの取得が終わったので、データベースとの接続を閉じています。使い終わった接続は必ず閉じることでリソースを解放します。

LINE 099: 履歴が空かどうかの判定。取得したrowsが空かどうかを確認しています。1件もデータがない場合は、次の行の特別な表示に切り替わります。

LINE 100: 履歴なし時の表示。受信履歴が1件もない場合に表示する案内メッセージのHTMLを返しています。テーブルの行として「まだ受信履歴はありません」と表示されます。

LINE 102: 行リストの初期化。HTMLの行を格納するための空のリストを用意しています。この後のループでここに各行の文字列が追加されていきます。

LINE 103: 取得データを1件ずつ処理。rowsに含まれる各レコードを1件ずつ取り出し、受信日時・イベント種別・ペイロード・状態の4つの変数に分けています。LINE 104: HTML行の組み立て開始。

1件分のデータをHTMLのテーブル行として組み立て、cellsリストへ追加する処理を開始しています。LINE 105: 行タグの開始。テーブルの1行を表す開始タグを記述しています。

この後に各データを表示するセルが続きます。LINE 106: 受信日時セルの生成。受信日時をエスケープ処理したうえでセルとして埋め込んでいます。

エスケープにより特殊文字が正しく表示されます。LINE 107: イベント種別セルの生成。イベント種別の値をエスケープしてセルに埋め込んでいます。

これにより安全にHTML上へ表示できます。LINE 108: ペイロードセルの生成。JSONペイロードをpreタグで整形しつつエスケープして表示しています。

改行や記号を保ったまま安全に画面へ出力します。LINE 109: 処理状態セルの生成。処理済やエラーといった状態文字列をエスケープしてセルに埋め込んでいます。

ユーザーは一目でイベントの処理結果を確認できます。LINE 110: 行タグの終了。テーブルの1行を閉じる終了タグを記述しています。

ここまでで1件分のHTML行が完成します。LINE 111: cellsへの追加終了。appendメソッドの呼び出しを閉じ、組み立てた1行分のHTML文字列をcellsリストに追加しています。

LINE 112: 全行を連結して返却。cellsリストに溜まった各行のHTML文字列をすべて連結し、1つの文字列として返しています。この結果がテーブル本体に挿入されます。

RUN 5/6: 履歴行のHTMLを組み立てる。render_rowsを呼び、保存済みイベントがHTMLの行に変換されることを確認します。CHECK 5/6: 途中実行に成功。

行HTMLの長さ: 211 先頭100文字: <tr><td>2026-07-30 06:17:13</td><td>ping</td><td><pre>{&quot;ok&quot;: 1}</pre></td><td>処理済</td></tr RETURN 05: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。

LINE 116: トップページのルート定義。ブラウザからルートパスへアクセスした際に呼び出されるエンドポイントを定義しています。レスポンスの形式としてHTMLを指定しています。

LINE 117: 履歴表示関数の定義。受信履歴ページを返す関数を定義しています。この関数の中でHTMLテンプレートを組み立てて返却します。

LINE 118: HTMLテンプレートの開始。画面全体のHTMLを1つの文字列として記述し始めています。doctype宣言からHTML文書の作成が始まります。

LINE 119: html要素の開始。ページ全体を囲むhtmlタグを開始し、言語設定を日本語に指定しています。LINE 120: head要素の開始。

ページのメタ情報やタイトル、スタイルを記述するhead部分を開始しています。LINE 121: 文字コードの指定。ページの文字コードをUTF-8に設定しています。

これにより日本語が文字化けせずに表示されます。LINE 122: ページタイトルの設定。ブラウザのタブに表示されるページタイトルを「Webhook受信履歴」に設定しています。

LINE 123: スタイル定義の開始。ページの見た目を整えるCSSスタイルの記述を開始しています。LINE 124: 全体のフォントと余白設定。

body要素にフォントや余白、文字色を指定しています。ページ全体の見やすさを整えるための設定です。LINE 125: ラベルの表示スタイル。

label要素をブロック表示にし、上余白と太字を設定しています。フォームの項目名が見やすくなります。LINE 126: 入力欄のサイズ設定。

inputとtextareaの幅や内側の余白を指定しています。入力しやすい大きさに整えています。LINE 127: ボタンの見た目設定。

button要素に上余白と内側の余白を設定しています。クリックしやすいボタンの見た目にしています。LINE 128: テーブル全体のスタイル設定。

テーブルの枠線を結合し、幅を100%にして上余白を設定しています。テーブル全体のレイアウトを整えています。LINE 129: セルの枠線と余白設定。

見出しとデータのセルに枠線と余白、文字寄せを指定しています。表の各セルが見やすく整います。LINE 130: preタグの表示調整。

pre要素の余白をなくし、長い文章でも折り返して表示されるように設定しています。LINE 131: 空状態メッセージのスタイル。履歴が空のときに表示するメッセージの文字色と中央寄せを設定しています。

LINE 132: スタイル定義の終了。CSSスタイルの記述を終了しています。LINE 133: head要素の終了。

head部分の記述を終了しています。ここまでがページのメタ情報部分です。LINE 134: body要素の開始。

実際に画面へ表示される本文部分の記述を開始しています。LINE 135: 見出しの表示。ページの見出しとして「Webhook受信アプリ」という文字を表示しています。

LINE 136: フォームの開始。Webhookを送信するためのフォームを開始しています。onsubmitでページの再読み込みを止める設定にしています。

LINE 137: イベント種別ラベルの表示。イベント種別の入力欄に対応するラベルを表示しています。LINE 138: イベント種別入力欄の設置。

イベント種別を入力するためのテキストボックスを配置しています。入力例としてorder.createdが表示されます。LINE 139: ペイロードラベルの表示。

JSONペイロードを入力する欄に対応するラベルを表示しています。LINE 140: ペイロード入力欄の設置。JSON形式のペイロードを入力するテキストエリアを配置しています。

入力例として簡単なJSONが表示されます。LINE 141: 改行の挿入。入力欄とボタンの間に改行を入れて、見た目を整えています。

LINE 142: 送信ボタンの設置。Webhookを送信するためのボタンを配置しています。クリックすると後述のJavaScript処理が実行されます。

LINE 143: フォームの終了。Webhook送信用フォームの記述を終了しています。LINE 144: 履歴テーブルの開始。

受信履歴を表示するテーブルの記述を開始しています。LINE 145: テーブルヘッダーの開始。テーブルの見出し部分の記述を開始しています。

LINE 146: 見出し行の表示。受信日時・イベント種別・JSONペイロード・処理状態という4つの見出しを表示しています。LINE 147: テーブルヘッダーの終了。

テーブルの見出し部分の記述を終了しています。LINE 148: テーブル本体の開始。実際のデータ行を表示するテーブル本体部分を開始しています。

LINE 149: 行データの差し込み位置。render_rows関数で作成された履歴行のHTMLが後で差し込まれる目印を配置しています。LINE 150: テーブル本体の終了。

テーブル本体部分の記述を終了しています。LINE 151: テーブルの終了。履歴を表示するテーブル全体の記述を終了しています。

LINE 152: スクリプトの開始。ボタン操作に対応するJavaScriptの記述を開始しています。LINE 153: 送信ボタンのイベント登録。

送信ボタンがクリックされたときに実行される非同期処理を登録しています。この処理でWebhookの送信が行われます。LINE 154: イベント種別の取得。

入力欄からイベント種別の値を取得し、未入力の場合はunknownを使うようにしています。LINE 155: ペイロードの取得。テキストエリアからJSONペイロードの値を取得し、未入力の場合は空のJSONを使うようにしています。

LINE 156: Webhookエンドポイントへの送信開始。取得したイベント種別をクエリパラメータに含めて、/webhookへリクエストを送る処理を開始しています。LINE 157: HTTPメソッドの指定。

送信するリクエストのメソッドをPOSTに指定しています。LINE 158: リクエストヘッダーの指定。送信するデータの形式がJSONであることをヘッダーで指定しています。

LINE 159: 送信データの指定。リクエストの本文として、入力されたペイロードをそのまま送信データに設定しています。LINE 160: fetch呼び出しの引数を閉じる。

fetchメソッドに渡していたオプションオブジェクトと呼び出し自体を閉じている行です。これによりWebhook送信のPOSTリクエストの記述が完成します。LINE 161: 0.8秒後にページを再読み込み。

送信処理が完了した後、0.8秒待ってからページを再読み込みするsetTimeoutの処理です。これにより保存された最新の受信履歴が画面に反映されます。LINE 162: クリックイベントの処理を閉じる。

送信ボタンのクリック時に実行する非同期関数と、addEventListenerの呼び出しを閉じている行です。ここまでがボタン操作の一連の処理となります。LINE 163: scriptタグを閉じる。

画面内に埋め込んでいたJavaScriptの記述範囲を終了させるための閉じタグです。これより後はHTML本体の終了部分になります。LINE 164: bodyタグを閉じる。

ページの本文領域を示すbodyタグを閉じている行です。フォームや履歴テーブルなど画面に表示する内容の記述がここで終わります。LINE 165: htmlタグとテンプレート文字列を閉じる。

ページ全体を囲むhtmlタグを閉じるとともに、Pythonの複数行文字列としてのテンプレートを終了させている行です。これでHTMLの雛形全体が完成します。LINE 166: 行データを埋め込んで応答を返す。

テンプレート中のプレースホルダーをrender_rows関数が作った履歴の行に置き換え、その結果をHTMLResponseとして返している行です。ブラウザにはここで完成した履歴ページが表示されます。RUN 6/6: 履歴ページ全体を表示する。

GET /を開き、履歴ページのHTMLが返ることを確認します。CHECK 6/6: 途中実行に成功。見出しあり: True 表の列見出し: True RETURN 06: エディターへ戻りました。

入力済みのコードを維持したまま、次の実装へ進みます。LINE 170: スクリプト直接実行時の判定。このファイルが直接実行されたときだけ以降の処理を行うようにする、Pythonでよく使われる書き方です。

他のファイルからインポートされた場合はここは実行されません。LINE 171: uvicornモジュールの読み込み。アプリを起動するためのASGIサーバーであるuvicornを、この時点で読み込んでいる行です。

開発用サーバーとして動かすときにのみ必要なため、ここでインポートしています。LINE 173: 開発用サーバーの起動。作成したFastAPIアプリをローカルホストのポート8000番で起動している行です。

この処理によりブラウザからWebhook受信履歴の画面にアクセスできるようになります。PREVIEW: 完成したWebhook受信アプリを操作。入力・送信・結果表示を実際のブラウザ画面で確認します。

UPLOAD: 実際の入力データでアプリを操作。フォームにイベント種別とJSONペイロードを入力してWebhookを送信すると、背景処理でSQLiteへ保存され、受信履歴の表に種別と処理状態(処理済)が表示されるのを確認できた 確認1/4: 受信前の履歴画面を開く。サーバー起動直後にGET /を開き、見出しと空の履歴メッセージが表示されることを確認します。

確認2/4: 妥当なJSONのWebhookを送信する。イベント種別を指定して正しいJSONをPOSTし、202応答と処理済の履歴を確認します。確認3/4: 壊れたJSONを送信する。

JSONとして解釈できない本文をPOSTし、処理状態がエラーとして記録されることを確認します。確認4/4: イベント種別の決まり方を確認する。ヘッダー指定と未指定の2件を送信し、ヘッダー由来の種別とunknownの既定値を確認します。

学習内容のまとめ。@app.postでPOST /webhookのエンドポイントを定義するsqlite3.connectでDBファイルへ接続する 生ボディを読み取りイベント種別と受信日時を決めるDBファイルをOSの一時ディレクトリ配下に置く 小さく実行確認しながら完成状態まで段階的に組み立てる エンディング。

Python研修はCodeCampでご確認ください。

PythonのFastAPIとSQLiteとは

今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。

WebフレームワークFastAPIの特徴

FastAPIは、Pythonで軽量なWeb APIやWebアプリを素早く構築できるWebフレームワークです。型ヒントを使ってリクエストの受け取りやレスポンス生成を簡潔に書ける点が魅力といえます。今回はPOSTでWebhookを受け取り、履歴ページを返す土台として活用します。

Webhook受信アプリでFastAPIが担う役割は次の機能に整理できます

  • @app.postでPOST /webhookのエンドポイントを定義する
  • @app.getでGET /の履歴ページを返す
  • Requestオブジェクトから生のボディを取得する
  • JSONResponseで202 Acceptedを返す
  • HTMLResponseで組み立てたHTMLを返す

受信履歴を保存するSQLiteの役割

SQLiteは、Python標準のsqlite3モジュールから扱える軽量な組み込みデータベースです。サーバーを別途立てずに1つのファイルへデータを保存でき、小さなアプリの永続化に向いています。本教材では受信したWebhookを1件ずつテーブルへ記録し、履歴表示の元データにします。

SQLiteへ受信イベントを保存するために用意している要素は次のとおりです

  • sqlite3.connectでDBファイルへ接続する
  • webhook_eventsテーブルへ4項目を保存する
  • received_atとevent_typeとpayloadとstatusの列を持つ
  • INSERT文で受信イベントを1件追加する
  • ORDER BY id DESCで新しい順に取り出す
さらに詳しく学べるPython研修の詳細はこちら

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]"
.\.venv\Scripts\python.exe -c "import sqlite3; print('SQLite', sqlite3.sqlite_version)"
.\.venv\Scripts\python.exe -c "from fastapi import BackgroundTasks; print('BackgroundTasks', BackgroundTasks)"

macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。

python3 -m venv .venv
./.venv/bin/python -m pip install --upgrade pip
./.venv/bin/python -m pip install fastapi "uvicorn[standard]"
./.venv/bin/python -c "import sqlite3; print('SQLite', sqlite3.sqlite_version)"
./.venv/bin/python -c "from fastapi import BackgroundTasks; print('BackgroundTasks', BackgroundTasks)"
  • SQLiteはPython標準ライブラリのsqlite3を使うため追加インストールは不要です。BackgroundTasksもfastapiに同梱されています。
  • 起動は.\.venv\Scripts\python.exe -m uvicorn webhook_app:app --reloadで、ブラウザからhttp://127.0.0.1:8000/ を開きます。
  • 初回アクセス時にwebhooks.dbが実行フォルダへ作られ、受信履歴が永続化されます。まっさらにしたいときはこのファイルを削除します。
さらに詳しく学べるPython研修の詳細はこちら

Webhook受信アプリの要件定義

目的は、FastAPIのPOSTエンドポイントでWebhookのJSONを受信し、SQLiteへ保存した受信履歴を一覧画面で確認できる小さなWebアプリを作ることです。

対象者として、PythonとHTTPの基礎を理解し、FastAPIでWebhookを受信して保存する実践的なWebアプリの作り方を学びたい人を想定しています。

完成物は、FastAPIとSQLiteとBackgroundTasksで作り、POSTでWebhookを受信し履歴画面へ一覧表示するWebhook受信アプリです。

実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。

機能要件

  • POST /webhookでWebhookのJSONペイロードを受信する
  • 受信直後に202 Acceptedを返す
  • BackgroundTasksで応答後にSQLiteへ1件保存する
  • イベント種別をクエリパラメータかヘッダーから取得する
  • 取得できないときイベント種別をunknownとする
  • 受信日時を年月日時分秒の文字列で記録する
  • ペイロードがJSONとして妥当なら処理済とする
  • JSONとして壊れていればエラーとする
  • GET /で受信履歴を新しい順に一覧表示する
  • 受信履歴が無いとき空メッセージを表示する

非機能要件

  • DBファイルをOSの一時ディレクトリ配下に置く
  • 起動時のlifespanでテーブルを初期化する
  • CREATE TABLE IF NOT EXISTSで再起動時も安全に作成する
  • HTML表示時にescapeでエスケープする
  • 追加のフロントエンドや外部DBを使わない自己完結構成
  • 空のリクエストボディを空JSONオブジェクトとして扱う
  • 保存前にJSONをキー順で正規化する

実装方針

今回はFastAPIとSQLiteの基本動作を追いやすくするため、Webhook受信アプリ本体を1つのPythonファイルへまとめます。

入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。

Webhook受信アプリを安全に組み立てるための実装方針は次のとおりです。

  • DBファイルをOSの一時ディレクトリ配下に置く
  • 起動時のlifespanでテーブルを初期化する
  • CREATE TABLE IF NOT EXISTSで再起動時も安全に作成する
  • HTML表示時にescapeでエスケープする
  • 追加のフロントエンドや外部DBを使わない自己完結構成
  • 空のリクエストボディを空JSONオブジェクトとして扱う
  • 保存前にJSONをキー順で正規化する

完成と判断する条件

  1. POST /webhookが202 Acceptedを返す
  2. 背景処理の後にSQLiteの受信イベントが1件増える
  3. 妥当なJSONの受信で処理状態が処理済になる
  4. 壊れたJSONの受信で処理状態がエラーになる
  5. GET /で受信履歴の表が新しい順に並ぶ
  6. 履歴が無いときまだ受信履歴はありませんと表示する
  7. イベント種別が未指定のときunknownと記録される
さらに詳しく学べるPython研修の詳細はこちら

FastAPIでWebhook受信アプリを作る際の重要ポイント

このアプリの中心は、Webhookを受け取ってすぐに応答し、時間のかかる保存処理を後回しにする流れです。FastAPIのBackgroundTasksへ保存関数を登録すると、レスポンスを返した後にSQLiteへの書き込みが実行されます。受け取った側を待たせずに履歴を蓄積できる点が、この設計の狙いになります。

Webhook受信から履歴表示までの処理は次の順序で進みます

  • 生ボディを読み取りイベント種別と受信日時を決める
  • BackgroundTasksへ保存関数を登録する
  • 202 Acceptedを先に返す
  • 応答後にJSON妥当性を判定して保存する
  • GET /で保存済みの履歴を表に描画する

背景処理で応答を待たせない仕組み

受信ハンドラはbackground_tasks.add_taskで保存関数と引数を登録し、すぐにレスポンスを返します。保存処理はレスポンス送信後に動くため、送信元は保存の完了を待たずに済みます。

BackgroundTasksへ登録している内容は次のとおりです

  • save_webhook_event_to_sqliteを登録する
  • 受信日時とイベント種別とペイロードを渡す

処理状態を判定する条件

保存関数はjson.loadsでペイロードを解析し、成功なら処理済、失敗ならエラーという状態を付けます。妥当なJSONはキー順に正規化してから保存し、履歴の表示を整えます。

処理状態が決まる条件は次のとおりです

  • JSONとして解析できれば処理済
  • 解析に失敗すればエラー
  • 正規化した本文を保存する

履歴をHTMLに描画する方法

履歴ページはrender_rowsでSELECT結果を1行ずつHTMLの表へ変換します。各セルはescapeでエスケープしてから埋め込み、テンプレートの__ROWS__を置き換えて返します。

履歴描画で行っている処理は次のとおりです

  • 新しい順に全件を取得する
  • 各値をescapeで安全に埋め込む
  • 履歴が無いとき空メッセージを出す

Python研修一覧はこちら

目的に合うPython研修を一覧形式から探したい方は、ぜひご利用ください。

Python研修を比較する

Java研修一覧はこちら

目的に合うJava研修を一覧形式から探したい方は、ぜひご利用ください。

Java研修を比較する

PHP研修一覧はこちら

目的に合うPHP研修を一覧形式から探したい方は、ぜひご利用ください。

PHP研修を比較する

新入社員研修

目的に合う新入社員研修を一覧形式から探したい方は、ぜひご利用ください。

新入社員研修を比較する

全ての研修からも探したい方はこちら

PythonでWebhook受信アプリの完成コード

コードは、DBを初期化するinit_db、保存を担うsave_webhook_event_to_sqlite、起動処理のlifespan、受信するreceive_webhook、履歴を組み立てるrender_rowsで構成しています。

起動時に一度だけテーブルを作り、受信のたびに保存処理を背景へ登録する流れです。DBの場所は書き込みできないことがある配置先を避け、OSの一時ディレクトリに置きました。

受信の入口では生の本文をそのまま読み取り、種別だけをクエリかヘッダーから拾います。保存の判定はJSONとして読めるかどうかで分け、状態を付けてINSERTします。

このセクションの用語

BackgroundTasks
FastAPIで、応答を返した後に実行したい処理を登録できる仕組みです。時間のかかる作業を後回しにできます。
lifespan
アプリの起動時と終了時に走らせたい処理を書ける仕組みで、起動時のテーブル作成などに使います。
プレースホルダ
SQL文の中で値の位置を?で示し、実際の値を別に渡す書き方です。値を直接埋め込む危険を避けられます。
デコレータ
関数の上に@付きで書き、その関数に機能を付け足す仕組みです。@app.postでルート登録に使います。
受信データがSQLiteへ保存されるまでのデータの流れ request.body()で生JSONを取得typeクエリかヘッダーで種別を決定add_taskで保存処理を背景へ登録json.loadsで妥当性を判定プレースホルダ付きINSERTで保存
受信データがSQLiteへ保存されるまでのデータの流れ
import json
import sqlite3
import tempfile
from contextlib import asynccontextmanager
from datetime import datetime
from html import escape
from pathlib import Path

from fastapi import BackgroundTasks, FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse

# SQLiteデータベースの場所と初期化
# 実行環境によっては配置ディレクトリが書き込み不可のことがあるため、
# 常に書き込み可能なOSの一時ディレクトリ配下にDBファイルを置く
DB_PATH = Path(tempfile.gettempdir()) / "webhooks.db"


def init_db() -> None:
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS webhook_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            received_at TEXT NOT NULL,
            event_type TEXT NOT NULL,
            payload TEXT NOT NULL,
            status TEXT NOT NULL
        )
        """
    )
    conn.commit()
    conn.close()


def save_webhook_event_to_sqlite(received_at: str, event_type: str, payload: str) -> None:
    """受信イベントを背景処理でSQLiteへ保存する。

    BackgroundTasksからこの関数が呼び出され、レスポンスを返した後に
    非同期でSQLiteへ1件INSERTする。JSONとして妥当なら「処理済」、
    壊れていれば「エラー」として保存する。
    """
    # ペイロードがJSONとして妥当かを判定して処理状態を決める
    try:
        parsed = json.loads(payload)
        normalized = json.dumps(parsed, ensure_ascii=False, sort_keys=True)
        status = "処理済"
    except (ValueError, TypeError):
        normalized = payload
        status = "エラー"

    # SQLiteへ1件挿入して永続化する(背景処理としてINSERTを実行)
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "INSERT INTO webhook_events (received_at, event_type, payload, status)"
        " VALUES (?, ?, ?, ?)",
        (received_at, event_type, normalized, status),
    )
    conn.commit()
    conn.close()


# アプリ起動時にテーブルを用意する
@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield


app = FastAPI(lifespan=lifespan)
ROUTES = ["/"]


# WebhookのJSONイベントをPOSTで受信する
@app.post("/webhook")
async def receive_webhook(request: Request, background_tasks: BackgroundTasks):
    raw = await request.body()
    payload = raw.decode("utf-8").strip() or "{}"
    event_type = (
        request.query_params.get("type")
        or request.headers.get("x-event-type")
        or "unknown"
    )
    received_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    # 受信イベントを背景処理でSQLiteへ保存する(BackgroundTasksへ登録)
    background_tasks.add_task(save_webhook_event_to_sqlite, received_at, event_type, payload)
    return JSONResponse({"status": "accepted"}, status_code=202)


# 受信履歴を読み出してHTMLの行を組み立てる
def render_rows() -> str:
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        "SELECT received_at, event_type, payload, status"
        " FROM webhook_events ORDER BY id DESC"
    ).fetchall()
    conn.close()

    if not rows:
        return '<tr><td colspan="4" class="empty">まだ受信履歴はありません</td></tr>'

    cells = []
    for received_at, event_type, payload, status in rows:
        cells.append(
            "<tr>"
            f"<td>{escape(received_at)}</td>"
            f"<td>{escape(event_type)}</td>"
            f"<td><pre>{escape(payload)}</pre></td>"
            f"<td>{escape(status)}</td>"
            "</tr>"
        )
    return "".join(cells)


# ブラウザでWebhook受信履歴を表示する
@app.get("/", response_class=HTMLResponse)
def history_page() -> HTMLResponse:
    template = """<!doctype html>
<html lang="ja">
<head>
<meta charset="utf-8">
<title>Webhook受信履歴</title>
<style>
 body { font-family: sans-serif; margin: 2rem; color: #222; }
 label { display: block; margin-top: .5rem; font-weight: bold; }
 input, textarea { width: 26rem; padding: .3rem; }
 button { margin-top: .8rem; padding: .4rem 1rem; }
 table { border-collapse: collapse; width: 100%; margin-top: 1.5rem; }
 th, td { border: 1px solid #ccc; padding: .4rem .6rem; text-align: left; vertical-align: top; }
 pre { margin: 0; white-space: pre-wrap; word-break: break-all; }
 .empty { color: #888; text-align: center; }
</style>
</head>
<body>
<h1>Webhook受信アプリ</h1>
<form id="webhook-form" onsubmit="return false;">
 <label for="event-type">イベント種別</label>
 <input id="event-type" name="event-type" placeholder="order.created">
 <label for="payload">JSONペイロード</label>
 <textarea id="payload" name="payload" rows="4" placeholder='{"id": 1, "amount": 1200}'></textarea>
 <br>
 <button id="send-webhook" type="button">Webhookを送信</button>
</form>
<table id="history-table">
 <thead>
  <tr><th>受信日時</th><th>イベント種別</th><th>JSONペイロード</th><th>処理状態</th></tr>
 </thead>
 <tbody id="history-body">
__ROWS__
 </tbody>
</table>
<script>
document.getElementById('send-webhook').addEventListener('click', async () => {
  const eventType = document.getElementById('event-type').value || 'unknown';
  const payload = document.getElementById('payload').value || '{}';
  await fetch('/webhook?type=' + encodeURIComponent(eventType), {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: payload
  });
  setTimeout(function () { window.location.reload(); }, 800);
});
</script>
</body>
</html>"""
    return HTMLResponse(template.replace("__ROWS__", render_rows()))


# 開発用サーバーの起動
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。

tempfileでDBの置き場所を決める

DB_PATH = Path(tempfile.gettempdir()) / "webhooks.db"

DBファイルをtempfile.gettempdir()が返す一時ディレクトリの下に置いています。配置ディレクトリが書き込み不可でも、ここなら安全に書き込めるので初心者がハマりにくいです。

init_dbでwebhook_eventsテーブルを作る

    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS webhook_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            received_at TEXT NOT NULL,
            event_type TEXT NOT NULL,
            payload TEXT NOT NULL,
            status TEXT NOT NULL
        )
        """
    )

CREATE TABLE IF NOT EXISTSで、受信日時・種別・ペイロード・状態を持つテーブルを用意します。すでに存在すれば作り直さないので、起動を繰り返しても安全です。

lifespanで起動時に一度だけ初期化する

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield

lifespanはアプリの起動と終了に合わせて動く仕組みで、yieldの手前が起動時の処理です。ここでinit_dbを呼び、テーブルが無ければ作成しておきます。

receive_webhookで本文と種別を受け取る

    raw = await request.body()
    payload = raw.decode("utf-8").strip() or "{}"
    event_type = (
        request.query_params.get("type")
        or request.headers.get("x-event-type")
        or "unknown"
    )

await request.body()で本文を生のまま読み、空なら{}で補います。種別はtypeクエリ、次にx-event-typeヘッダーの順で探し、どちらも無ければunknownにします。

add_taskで保存を背景処理へ登録する

    background_tasks.add_task(save_webhook_event_to_sqlite, received_at, event_type, payload)
    return JSONResponse({"status": "accepted"}, status_code=202)

background_tasks.add_taskに保存関数と引数を渡すと、応答を返した後にその関数が実行されます。保存を待たずに202をすぐ返すので、送信側を長く待たせません。

json.loadsで処理状態を振り分ける

    try:
        parsed = json.loads(payload)
        normalized = json.dumps(parsed, ensure_ascii=False, sort_keys=True)
        status = "処理済"
    except (ValueError, TypeError):
        normalized = payload
        status = "エラー"

json.loadsで本文を読めれば整形して「処理済」に、読めず例外が出れば元の文字列のまま「エラー」にします。妥当性の判定と状態付けをここでまとめて行います。

プレースホルダ付きINSERTで1件保存する

    conn.execute(
        "INSERT INTO webhook_events (received_at, event_type, payload, status)"
        " VALUES (?, ?, ?, ?)",
        (received_at, event_type, normalized, status),
    )

値を?のプレースホルダで渡し、実データはタプルで別に渡しています。文字列を直接連結しない書き方なので、値に記号が混ざっても安全に1件挿入できます。

参考:

©FastAPI公式ドキュメントBackground Tasks

You can define background tasks to be run after returning a response.


Python基礎・実践(Django)

企業・法人向けのPython研修では、基礎から応用まで体系的に学べます。

Python研修の詳細

DX社員研修

企業・法人向けのDX研修では、実務に繋がるリスキリングでITレベルを向上させます。

DX研修の詳細

Javaエンジニア育成研修

企業・法人向けのJavaエンジニア育成研修では、Javaの基礎から応用まで確実に習得できます。

Java研修の詳細

新卒・新入社員向け研修

企業・法人に新入社員・新卒社員に向けたプログラミング研修を提供しています。

新入社員研修の詳細

コードキャンプのIT研修を全て見る

Webhook受信アプリの動作確認

実際にブラウザで履歴画面を開き、フォームにイベント種別とJSONペイロードを入力して送信しました。送信はローカルのサーバーとの通信だけで、外部通信なしで完結します。

送信するとフォームはfetch/webhookを呼び、少し待ってからページが再読み込みされました。背景処理でSQLiteへ保存され、受信履歴の表に種別と処理状態(処理済)が表示されるのを確認できました。

操作前の画面はHTTP 200で表示されており、送信後は同じ表に新しい行が加わっています。受信日時・イベント種別・JSONペイロード・処理状態が横並びで読み取れます。

このセクションの用語

fetch
ブラウザ側からJavaScriptでサーバーへHTTPリクエストを送るための標準の関数です。
202 Accepted
リクエストを受け付けたが処理はまだ完了していない、という意味のHTTPステータスコードです。

画面上で確認できた挙動は次のとおりです。

  1. 操作前は履歴が空の初期画面が表示された(HTTP 200)
  2. フォームに種別とJSONペイロードを入力してWebhookを送信した
  3. 少し待つとページが再読み込みされ、履歴の表に反映された
  4. 表の行に種別と処理状態(処理済)が並んで表示された
操作前の画面: / (HTTP 200)
操作前の画面: / (HTTP 200)
フォームにイベント種別とJSONペイロードを入力してWebhookを送信すると、背景処理でSQLiteへ保存され、受信履歴の表に種別と処理状態(処理済)が表示されるのを確認できた
フォームにイベント種別とJSONペイロードを入力してWebhookを送信すると、背景処理でSQLiteへ保存され、受信履歴の表に種別と処理状態(処理済)が表示されるのを確認できた
実ブラウザ操作: フォームにイベント種別とJSONペイロードを入力してWebhookを送信すると、背景処理でSQLiteへ保存され、受信履歴の表に種別と処理状態(処理済)が表示されるのを確認できた
実ブラウザ操作: フォームにイベント種別とJSONペイロードを入力してWebhookを送信すると、背景処理でSQLiteへ保存され、受信履歴の表に種別と処理状態(処理済)が表示されるのを確認できた
さらに詳しく学べるPython研修の詳細はこちら

PythonのFastAPIとSQLite利用時に起きやすいエラーと対処法

ここでは初心者が遭遇しやすい実行時のエラーを整理します。多くは環境の準備不足か、送信データの形に原因があります。

エラー表示に慌てず、メッセージの中の語句を手がかりに原因を切り分けると解決が早いです。

このセクションの用語

仮想環境
プロジェクトごとにライブラリを分けて入れられる、独立したPython環境です。
エラー例 原因 対処
sqlite3.OperationalError: unable to open database file DBの配置先が書き込み不可 tempfile.gettempdir()配下など書き込み可能な場所にDB_PATHを置く
ModuleNotFoundError: No module named 'fastapi' FastAPIが未インストールか仮想環境が未有効 仮想環境を有効化してfastapiを入れ直す
イベント種別がunknownになる typeクエリもX-Event-Typeヘッダーも付けていない 送信時にtypeをクエリかヘッダーで渡す
履歴の処理状態がエラーになる ペイロードがJSONとして壊れている json.loadsが通る正しいJSON文字列を送る
OSError: address already in use 起動ポートが既に使用中 別ポートで起動するか、使用中のプロセスを止める

Webhook受信アプリで注意したい点

SQLiteはファイル1つで完結し、sqlite3.connectで接続して読み書きします。変更はconn.commit()で確定し、最後にconn.close()で閉じる素直な書き方にしました。

つまずきやすいのは保存のタイミングです。BackgroundTasksは応答を返した後に走る非同期処理なので、送信直後に履歴を見ると、まだ反映されていないことがあります。

だから送信後は少し待ってから再読み込みする作りにしました。DBファイルの置き場所が書き込み可能かも、最初に確認しておくと安心です。

ポイントとしては、応答をすぐ返す背景保存の設計と、書き込み可能な場所へのDB配置を押さえておくことです。

POINT

DB配置:書き込み可能な場所へ

背景保存:応答後に非同期でINSERT

反映待ち:送信後は少し待って再読み込み

さらに詳しく学べるPython研修の詳細はこちら

Webhook受信アプリを実務で使う場面

この受信アプリは、外から届くイベントをいったん受け止めて、あとから中身と結果を見返す用途に向いています。処理状態で「処理済」と「エラー」を仕分けられるのも便利です。

実務でそのまま、あるいは少し拡張して使える場面を挙げます。

このセクションの用語

SaaS
インターネット経由で利用するソフトウェアの提供形態です。多くがWebhook通知に対応します。
使える場面 具体的な使い方
外部SaaSからの通知受信 決済やコード管理サービスのWebhookを受け取り、種別ごとに履歴へ残して後追いする
送信側の開発とテスト 実装した送信処理が正しく届くか、ペイロードと処理状態を表で目視確認する
簡易な受信ログの保管 受信日時・種別・JSONを一覧化し、障害調査で何がいつ届いたか振り返る
社内システムのイベント集約 複数システムからのPOSTを一箇所に集め、処理済とエラーで仕分ける
さらに詳しく学べるPython研修の詳細はこちら

Webhook受信アプリ開発のまとめ

FastAPISQLiteBackgroundTasksを組み合わせ、Webhookを受信して履歴で見返せる自己完結アプリを作りました。

POST /webhookで受信し、応答をすぐ返してから背景処理でSQLiteへ保存し、GET /の表で受信日時・種別・ペイロード・状態を一覧する流れです。

実際にフォームから送信すると、背景処理で保存され、履歴の表に種別と処理状態(処理済)が表示されるのを確認できました。

次は種別ごとの絞り込みや、届いたイベントに応じた自動処理を足すと、さらに実用的になります。

さらに詳しく学べるPython研修の詳細はこちら

参考にした一次情報

  1. ^ FastAPI - Background Tasks. https://fastapi.tiangolo.com/tutorial/background-tasks/, (参照26-07-30).
  2. ^ FastAPI - Using the Request Directly. https://fastapi.tiangolo.com/advanced/using-request-directly/, (参照26-07-30).
  3. ^ Python標準ライブラリsqlite3. https://docs.python.org/ja/3/library/sqlite3.html, (参照26-07-30).
  4. ^ Uvicorn - ASGI web server. https://www.uvicorn.org/, (参照26-07-30).

※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。

ITやプログラミングに関するコラム


ITやプログラミングに関するニュース

ブログに戻る

コメントを残す

コメントは公開前に承認される必要があることにご注意ください。

企業・法人向けのIT・プログラミング・生成AI研修を探す、比較する - IT・プログラミングを知って学べるコネクトメディア CodeCampが提供するDX人材育成が可能なプログラミングやITが学べる公開講座 - IT・プログラミングを知って学べるコネクトメディア コードキャンプが提供する無料で学べるプログラミングスクール講座 - IT・プログラミングを知って学べるコネクトメディア コードキャンプDX人材育成研修 - IT・プログラミングを知って学べるコネクトメディア 3.5日の研修で、年間1,600時間の削減効果が見込まれる。東京きらぼしフィナンシャルグループのDX人材育成事例 - IT・プログラミングを知って学べるコネクトメディア 配属3ヶ月で30%の生産性向上を実現するいよぎんコンピュータサービスの新人研修に迫る - IT・プログラミングを知って学べるコネクトメディア 金融業界の業務効率化を加速するニッセイアセットマネジメントの生成AI×GAS活用研修事例 - IT・プログラミングを知って学べるコネクトメディア 【製造業のDX人材育成事例】デジタル人材の即戦力化を実現する、日本ガイシ株式会社の異動者向オンボーディング研修 - ITやプログラミングを知って学べるコネクトメディア フューチャーアーキテクト株式会社が実現した新入社員向けIT研修プログラムでタスクフォース制度が主体的な学びと成長を生み出す - IT・プログラミングを知って学べるコネクトメディア コードキャンプDX人材育成研修 - IT・プログラミングを知って学べるコネクトメディア コードキャンプIT・プログラミング研修事例/【IT新入社員研修】オンラインとオフラインの最適バランスを実現したFutureOneの導入事例 - IT・プログラミングを知って学べるコネクトメディア コードキャンプIT・プログラミング研修事例/【新入社員研修】柔軟なハイブリッド型Java研修で実現した新卒20名の成長と成果|サークレイス株式会社 - ITやプログラミングを知って学べるコネクトメディア コードキャンプIT・プログラミング研修事例/現場により近いところにデジタルを根付かせるDX基礎講座研修|株式会社ブリヂストン - ITやプログラミングを知って学べるコネクトメディア コードキャンプIT・プログラミング研修事例/業務の効率化・DX推進に向けたIT人材育成への第一歩|株式会社カナエ - ITやプログラミングを知って学べるコネクトメディア 企業・法人向けのIT・プログラミング研修 - ITやプログラミングを知って学べるコネクトメディア

新着記事

対象者別で探す

子供(小学生・中学生・高校生)向け
プログラミング教室検索する

子供(小学生・中学生・高校生)がロボットやプログラミング言語を学ぶことができるオフラインからオンラインスクールを検索、比較することが可能です。

子供(小学生・中学生・高校生)
プログラミング教室検索する

ITやプログラムなどの
最新情報を検索する

日々、新しいITやプログラミング言語の情報が流れていきますが、特定の情報を時系列でニュースやコラムを確認することができます。

ITやプログラムなどの
最新情報を検索する