🦖 第十一章:FastAPI巨獣の覚醒!万人打刻サーバー構築
前回の「Line打刻Web(LIFF)」コースでは、超クールな打刻ボタンを作成し、Line発行の安全認証(JWT Token)を取得した。 しかしフロントエンドは「ウェイター」のようなもの。注文を取るだけだ。実際にデータベースへ記録し、給与計算や遅刻判定を行う「厨房の裏方」こそがバックエンドサーバー(Backend Server)である。
従来のNode.js(Express)やPHPでバックエンドを構築した場合、朝9時に5000人の社員が同時打刻すると、サーバーは「ブロッキング」でクラッシュ。1000人の打刻が失敗し、会社全体が大混乱する。
この問題を解決するため、本章ではPython界で最も注目され、NodeJSやGoに匹敵する性能を誇るFastAPIを導入する。 その名の通り超高速、非同期(Async)をネイティブサポート、APIドキュメントも自動生成。 Vibe CodingでAIを指揮し、万人の同時接続に耐える巨獣級バックエンドを構築しよう!
⚡ 実戦1:FastAPIスケルトンの超速構築とngrokトンネリング
Lineアプリ開発(Line BotやLIFF連携)で最初に直面する壁は:
Lineサーバーは「https対応の公開サーバー」にしかデータを送信できないこと。
localhost:8000でローカル実行しても、Lineから見つからない!
ここで必要となるのがFastAPI(ローカルサーバー構築)とngrok(ローカルサーバーを公開URL化)の二大ツールだ。
💡 Vibe Prompt 実戦1:AIに最簡潔な起動スクリプトを書かせる
uvicornのパラメータを暗記する必要はない。AIに即実行可能なスケルトンを生成させよう。
[!IMPORTANT] 以下のPromptをAIに送信:
PythonでLine Botのバックエンドサーバーを開発中です。FastAPIで基本的なスケルトンプログラム(main.py)を作成してください。要件:1. FastAPIインスタンスappを作成2. ルートディレクトリ(GET /)で{"status": "ok", "message": "Punch System is running"}を返す3. Webhook受信用ルート(POST /api/webhook)で任意のJSONデータを受信しコンソールに表示4. サーバー起動用uvicornコマンド(port 8000、reload有効)5. 対応するrequirements.txt内容
🤖 AI生成の超速スターター:
# main.py
from fastapi import FastAPI, Request
# FastAPI巨獣インスタンス生成
app = FastAPI(title="Line Punch System API")
@app.get("/")
async def root():
"""サーバー生存確認用ヘルスチェック"""
return {"status": "ok", "message": "Punch System is running 🚀"}
@app.post("/api/webhook")
async def receive_webhook(request: Request):
"""
フロントエンドやLineサーバーからのデータを受信する「開放ゲート」
"""
# 非同期で受信JSONを読み取り
payload = await request.json()
print("📥 新規Webhookデータ受信:", payload)
# FastAPIは自動で辞書をJSONに変換
return {"status": "success", "received_data": payload}
起動コマンドとトンネリング:
- ターミナルで:
uvicorn main:app --reload --port 8000 - 第二ターミナルで:
ngrok http 8000 - ngrokが生成するURL(例:
https://1234-abcd.ngrok-free.app)をLine Developerコンソールに登録。これでPCがクラウドサーバーに!
🔐 実戦2:セキュリティ最前線(CORS設定)
ngrok URLをLIFFページに設定し「打刻」ボタンを押すと、ブラウザコンソールに真っ赤なエラー:
Blocked by CORS policy(クロスオリジンポリシーによりブロック)が表示される。
現代のブラウザは厳格なセキュリティ機構を持つ: 「許可なくして、Aサイト(Vercelのフロント)からBサイト(ngrokバックエンド)へのAPI呼び出しは禁止」 FastAPIに「警備員」を設置し「Vercelは味方だから通せ」と伝えよう。
💡 Vibe Prompt 実戦2:完璧なCORS Middleware設定
[!IMPORTANT] 以下のPromptをAIに送信:
フロントエンド(https://my-punch.vercel.app と http://localhost:5173)からFastAPIサーバーへAPI呼び出し時、CORSエラーが発生します。main.pyを修正し、CORSMiddlewareを追加してください。要件:1. 上記2オリジンを許可2. 全HTTPメソッド(GET,POST,OPTIONS等)を許可3. 全ヘッダーを許可CORS設定を含む完全なFastAPI初期化コードを生成してください。
🤖 AIの企業級セキュリティ設定:
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(title="Line Punch System API")
# 「ホワイトリスト」:これらのオリジンからのリクエストのみ許可
# ⚠️ 本番環境で["*"]は絶対禁止!無防備になる
origins = [
"http://localhost:5173", # ローカル開発用
"https://my-punch.vercel.app", # 本番フロントエンド
]
# クロスオリジン警備員(Middleware)設置
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # ホワイトリストのみ許可
allow_credentials=True, # Cookie認証許可
allow_methods=["*"], # 全HTTPメソッド許可
allow_headers=["*"], # 全カスタムヘッダー許可(後述のAuthorization含む)
)
# ... 以下に@app.get("/")ルートを継続 ...
🕵️♂️ 実戦3:LIFFのデジタル署名を解読(JWT検証)
フロントエンドコースで学んだ通り:
「フロントから送られるデータは一切信用するな」
もしフロントが{"user_name": "王社長", "action": "clock_in"}を送信しても、ハッカーは{"user_name": "李さん"}に改竄可能。
そこでフロントは乱碼**JWT Token(idToken)**を送信。
バックエンドはこのTokenを直接解読せず、Line公式サーバーに問い合わせる:
「この証明書は本物か?誰の発行か?」
💡 Vibe Prompt 実戦3:Line公式APIを呼び出しToken検証
サードパーティAPI連携は複雑で、公式ドキュメントだけではミスが起きやすい。
[!IMPORTANT] 以下のPromptをAIに送信:
FastAPIで、フロントエンドがHTTPヘッダーのAuthorizationにLine LIFF生成のJWT(Bearer <id_token>)を送信します。依存性注入関数verify_line_token(request: Request)を作成してください。処理内容:1. request.headersからAuthorizationを取得しtoken抽出2. tokenなしならHTTPException 401を発生3. httpxでLine検証API(https://api.line.me/oauth2/v2.1/verify)にPOSTリクエスト4. パラメータにid_tokenと環境変数LINE_LIFF_ID(client_id)を設定5. Lineからエラー返答時はHTTPException 401で拒否6. 検証成功時はユーザー名とline_id(sub)を含む辞書を返却
🤖 AIの究極偽造防止システム:
import os
import httpx
from fastapi import Request, HTTPException, status, Depends
async def verify_line_token(request: Request):
"""
この関数は会社の「虹彩スキャナー」。
全てのAPIリクエストは、この関数の検査を通過しなければ���らない!
"""
# 1. ヘッダーから通行証を抽出
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="セキュリティトークンが不足または形式不正"
)
# "Bearer "以降の乱碼を抽出
id_token = auth_header.split(" ")[1]
client_id = os.getenv("LINE_LIFF_ID")
# 2. この通行証をLine警察本部へ「本物か?」と照会
# 非同期httpxを使用(従来のrequestsはサーバーをブロック)
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.line.me/oauth2/v2.1/verify",
data={
"id_token": id_token,
"client_id": client_id
}
)
payload = response.json()
# 3. 検証結果判定
if response.status_code != 200 or "error" in payload:
# Lineが偽物と判定した場合、即座にハッカーを撃退!
print("❌ Token検証失敗:", payload.get("error_description"))
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="証明書が無効です。再ログインしてください"
)
# 4. 検証成功!Lineサーバーが本人の真正身分を教えてくれる
# payloadにはsub(Lineユーザー固有ID)、name(氏名)、picture(画像)が含まれる
print(f"✅ 認証完了:{payload.get('name')}さん、ようこそ!")
return {
"line_id": payload.get("sub"),
"name": payload.get("name"),
"picture": payload.get("picture")
}
🛡️ 実戦4:鉄壁の打刻API構築(Dependency連携)
虹彩スキャナー(verify_line_token)を手に入れた。これを打刻機の前に設置する。
FastAPIでは**Dependency Injection(依存性注入)**と呼び、その強力でエレガントな理由だ。
💡 Vibe Prompt 実戦4:検証と打刻ルートの連携
[!IMPORTANT] 以下のPromptをAIに送信:
打刻APIエンドポイントPOST /api/punchを作成してください。要件:1. FastAPIのDependsでverify_line_tokenをcurrent_userとして注入2. PydanticのBaseModelで受信データ形式を定義(action: strで'in'か'out'のみ許可)3. リクエスト受信時:"[{current_user['name']}] が打刻実行:{action}"と表示4. 成功時JSONを返却(ユーザー名と打刻時刻を含む)
🤖 AIの依存性注入を完璧に連携した打刻ルート:
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from datetime import datetime
# Pydanticデータモデル:第二の防護壁、フロントからのデータ形式チェック
class PunchRequest(BaseModel):
action: str # 'in'は出勤、'out'は退勤
# ルーターを独立管理可能
router = APIRouter()
# 🚀 究極にセキュアな打刻エンドポイント
@router.post("/api/punch")
async def handle_punch(
# ここで二つの処理が発生:
# 1. フロントからのbody形式チェック(PunchRequest)
request_data: PunchRequest,
# 2. 🔥 神級防御:このリクエストは虹彩スキャナー(verify_line_token)必須!
# スキャン失敗時はここで即座にエラーになり、以降の処理には進めない
# スキャン成功時、current_userにユーザーの実名とIDが格納される
current_user: dict = Depends(verify_line_token)
):
action_type = request_data.action
user_name = current_user["name"]
line_id = current_user["line_id"]
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 打刻行動を記録
# 実際のシステムでは次章でSupabaseへ記録する
print(f"🎯 [打刻成功] {user_name} ({line_id}) が{action_type}を実行。時刻:{now_str}")
return {
"status": "success",
"message": f"打刻成功!{user_name}さん、おはようございます���",
"punch_time": now_str,
"action": action_type
}
🔍 深層解析:
これが企業級バックエンド設計の真髄:「責任分離」。
従来のPHPでは、/api/punch内にToken検証100行、形式チェック50行、打刻処理...と続き、コードが汚くなる。
FastAPIの設計では:
PunchRequestが形式エラーをブロックDepends(verify_line_token)が不正アクセスをブロック- メイン関数
handle_punchは打刻記録のみに集中 これによりコードの可読性が向上。「休暇API」追加時もDepends(verify_line_token)を追加するだけで同等のセキュリティが得られる!
🗄️ 実戦5:SupabaseクラウドDB連携(Async書き込み)
ユーザー認証が完了したので、打刻記録をデータベースへ書き込む。
フロントエンドで使用したsupabase-jsとは異なり、Pythonではsupabase-pyを使用。FastAPIの非同期構造を活かし、DB接続が他ユーザーの打刻をブロックしないよう注意。
💡 Vibe Prompt 実戦5:安全高效なDB書き込み関数の作成
[!IMPORTANT] 以下のPromptをAIに送信:
FastAPIで打刻記録をSupabaseの"punch_logs"テーブルへ書き込みます。Supabaseクライアント初期化コードを生成してください(os.getenvでurlとkeyを取得)。次に、async関数record_punch_to_db(line_id: str, action: str)を作成。処理内容:1. line_id, action, 現在時刻をSupabaseへ書き込み2. supabase.table("punch_logs").insert(...).execute()を使用3. try...exceptで完全保護。DB接続失敗時もAPIがクラッシュしないようエラーメッセージを表示
🤖 AIの高級非同期DBモジュール:
import os
from supabase import create_client, Client
from datetime import datetime
# 1. Supabase接続エンジン作成
# (注:Python版supabase-pyは現在同期処理だが、async関数内で使用またはpostgrest-pyのasync版を使用可能)
url: str = os.getenv("SUPABASE_URL")
key: str = os.getenv("SUPABASE_SERVICE_KEY") # ⚠️ バックエンドは特権KeyでRLSを回避可能
try:
supabase_db: Client = create_client(url, key)
print("✅ Supabaseデータベース接続成功!")
except Exception as e:
print("❌ データベース接続失敗。環境変数を確認:", e)
# 2. 防呆書き込み関数
async def record_punch_to_db(line_id: str, action: str) -> bool:
"""
打刻記録を安全にクラウドDBへ書き込み。
True返却は成功、Falseは失敗を示す。