FastAPI × Next.js 14で作る機械学習Webアプリ【フルスタック開発】

はじめに

日々の行動と気分の記録から最適な行動をレコメンドする行動活性化療法をサポートするWebアプリケーションを、FastAPI（Python）とNext.js 14（TypeScript）を使ったフルスタック構成で開発しました。現在、機械学習・バックエンド・フルスタックエンジニアとして就職・転職活動中です。

本記事では、以下の内容について解説します：

• 技術選定の理由と根拠
• フルスタックアーキテクチャの設計
• 機械学習の組み込み方
• 92.6%のテスト成功率を達成したテスト戦略
• 実装で直面した課題と解決策

【こんな方におすすめ】
✅ フルスタック開発に興味がある
✅ FastAPIとNext.jsの実践例を知りたい
✅ 機械学習をWebアプリに組み込む方法を学びたい
✅ テスト駆動開発の実例を見たい

【GitHub】 https://github.com/Datarchpy/action-activation-therapy

---

プロジェクト概要

アプリケーションの目的

行動活性化療法（Behavioral Activation）は、うつ病治療に効果的な認知行動療法の一種です。
本アプリは、ユーザーが日々の行動と気分を記録し、機械学習で気分低下の予兆を検知、
パーソナライズされた行動提案を行います。

主要機能

• 📊 行動・気分記録: 5-10秒で完了する簡単入力
• 📈 データ可視化: 気分・エネルギー推移をグラフ表示
• 🤖 ML予兆検知: 気分低下パターンを機械学習で検知
• 💡 行動推薦: 過去データから効果的な行動を提案
• 📱 PWA対応: スマホアプリのようなUX
• 🔒 プライバシー重視: Firebase認証でセキュアなデータ管理

技術スタック

レイヤー	技術
フロントエンド	Next.js 14 (App Router), TypeScript, Tailwind CSS
バックエンド	FastAPI, SQLAlchemy, SQLite → PostgreSQL
機械学習	scikit-learn, pandas, numpy
データ可視化	Recharts
認証	Firebase Authentication
インフラ	Vercel (Frontend), Render (Backend)
CI/CD	GitHub Actions, pytest, Jest

プロジェクト規模

• バックエンド: 27テスト（92.6%成功）、コアAPI 100%カバレッジ
• フロントエンド: 30テスト（73.3%成功）
• 開発期間: 約4週間（設計・実装・テスト）
• 総コード行数: バックエンド約1,500行、フロントエンド約2,000行

---

なぜこの技術スタックを選んだのか

バックエンド: FastAPI を選んだ理由

✅ 機械学習との親和性

Pythonベースなので、scikit-learnやpandasとシームレスに統合できます。

実装例:

# /predict エンドポイントでMLモデルを直接呼び出し
@router.post("/predict/mood-risk")
async def predict_mood_risk(user_id: str, db: Session):
    # データ取得
    logs = get_recent_logs(user_id, days=7, db=db)
    features = extract_features(logs)

    # ML予測
    model = load_model()
    risk_score = model.predict_proba(features)[0][1]

    return {
        "risk_level": "high" if risk_score > 0.7 else "low",
        "confidence": risk_score
    }

✅ 高速な開発速度

• 自動生成されるOpenAPIドキュメント: Swagger UIで即座にAPI仕様を確認
• Pydanticによる型安全なバリデーション: データ検証が自動化
• 非同期処理のネイティブサポート: async/awaitで高速レスポンス

# Pydanticで自動バリデーション
class ActionLogCreate(BaseModel):
    user_id: str
    action_category: str
    mood_score: int = Field(..., ge=0, le=10, description="気分スコア (0-10)")
    energy_score: int = Field(..., ge=0, le=10)
    focus_score: int = Field(..., ge=0, le=10)

✅ パフォーマンス

• Node.jsに匹敵する高速性（Starlette/Uvicorn基盤）
• 非同期DB操作でスループット向上
• 実測で平均レスポンス時間50ms以下を達成

---

フロントエンド: Next.js 14 を選んだ理由

✅ App Routerの採用

• Server Componentsでパフォーマンス最適化
• React Server Actionsでフォーム処理を簡潔に
• ファイルベースルーティングで直感的な開発

// app/dashboard/page.tsx (Server Component)
export default async function DashboardPage() {
  // サーバーサイドでデータ取得（クライアント側のJSバンドルに含まれない）
  const logs = await getActionLogs(userId)

  return <Dashboard initialData={logs} />
}

✅ PWA対応の容易さ

// next.config.js
const withPWA = require('next-pwa')({
  dest: 'public',
  register: true,
  skipWaiting: true,
})

module.exports = withPWA({
  // Next.js設定
})

✅ TypeScriptとの相性

エンドツーエンドの型安全性を実現。APIレスポンスからUIまで一貫した型定義。

// 型安全なAPI呼び出し
interface ActionLog {
  id: number
  mood_score: number
  energy_score: number
  timestamp: string
}

async function getActionLogs(userId: string): Promise<ActionLog[]> {
  const response = await fetch(`${API_URL}/actions/${userId}`)
  return response.json()
}

---

なぜマイクロサービスではなくモノリスなのか

理由:

1. 開発速度の優先: MVPフェーズでは1人での開発効率を重視
2. オーバーヘッドの削減: サービス間通信のレイテンシを回避
3. 段階的な移行: 将来的にML予測部分だけ分離可能な設計

アーキテクチャ:

[Next.js Frontend] → REST API → [FastAPI Backend]
                                      ↓
                                 [ML Module]
                                      ↓
                                 [SQLite DB]

MLモジュールは独立したPythonパッケージとして設計しており、
将来的には別サービスとして切り出し可能です。

---

システムアーキテクチャ

レイヤード・アーキテクチャ

プロジェクト構成:

frontend/src/
  ├── app/              # App Router (Pages)
  ├── components/       # UIコンポーネント
  ├── lib/              # ユーティリティ・API呼び出し
  └── hooks/            # カスタムフック

backend/
  ├── app/
  │   ├── api/          # APIエンドポイント
  │   │   ├── actions.py      # 行動ログAPI
  │   │   ├── predictions.py  # ML予測API
  │   │   └── schemas.py      # Pydanticスキーマ
  │   ├── models/       # SQLAlchemyモデル
  │   │   ├── action_log.py
  │   │   └── prediction.py
  │   ├── core/         # 設定・DB接続
  │   │   ├── config.py
  │   │   └── database.py
  │   └── ml/           # 機械学習モジュール
  │       ├── feature_engineering.py
  │       └── model_loader.py
  └── tests/            # pytest テストスイート
      ├── test_api/
      ├── test_models/
      └── test_integration/

責務の分離:

• API層: リクエスト処理、バリデーション
• モデル層: データベーススキーマ、ビジネスロジック
• ML層: 特徴量エンジニアリング、予測処理
• Core層: 設定、DB接続など横断的関心事

---

データフロー

1. ユーザー操作 → API呼び出し

// frontend/src/lib/api.ts
export async function createActionLog(data: ActionLogData) {
  const response = await fetch(`${API_URL}/api/v1/actions/`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`
    },
    body: JSON.stringify(data),
  })

  if (!response.ok) {
    throw new Error('Failed to create action log')
  }

  return response.json()
}

2. FastAPI → DB操作とビジネスロジック

# backend/app/api/actions.py
@router.post("/actions/", response_model=ActionLogResponse)
async def create_action_log(
    action: ActionLogCreate,
    db: Session = Depends(get_db)
):
    # 前回ログ取得
    last_log = db.query(ActionLog).filter(
        ActionLog.user_id == action.user_id
    ).order_by(ActionLog.id.desc()).first()

    # 新規ログ作成
    db_action = ActionLog(**action.dict())

    # スコア差分計算（ビジネスロジック）
    if last_log:
        db_action.mood_score_diff = action.mood_score - last_log.mood_score
        db_action.energy_score_diff = action.energy_score - last_log.energy_score

    db.add(db_action)
    db.commit()
    db.refresh(db_action)

    return db_action

3. 統計・分析エンドポイント

@router.get("/actions/{user_id}/stats")
async def get_user_stats(
    user_id: str,
    days: int = 7,
    db: Session = Depends(get_db)
):
    # 指定期間のログ取得
    cutoff_date = datetime.now() - timedelta(days=days)
    logs = db.query(ActionLog).filter(
        ActionLog.user_id == user_id,
        ActionLog.timestamp >= cutoff_date
    ).all()

    if not logs:
        return {"total_logs": 0, "avg_mood": None}

    # 統計計算
    mood_scores = [log.mood_score for log in logs]
    return {
        "total_logs": len(logs),
        "avg_mood": sum(mood_scores) / len(mood_scores),
        "avg_energy": sum(log.energy_score for log in logs) / len(logs),
        "mood_trend": calculate_trend(mood_scores)
    }

---

機械学習パイプライン

学習フェーズ（オフライン処理）

# ml/training/mood_prediction.py
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

# 過去データから学習
df = load_action_logs()

# 特徴量エンジニアリング
X = df[['mood_score', 'energy_score', 'focus_score', 'time_of_day', 'day_of_week']]
y = df['mood_declined']  # 次回気分が下がったか（ラベル）

# 訓練・検証分割
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# モデル学習
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# 精度評価
accuracy = model.score(X_test, y_test)
print(f"Model accuracy: {accuracy:.2%}")

# モデル保存
import joblib
joblib.dump(model, 'models/mood_predictor.pkl')

推論フェーズ（オンライン処理）

# backend/app/api/predictions.py
from functools import lru_cache
from datetime import datetime, timedelta

# モデルキャッシング
_model_cache = {"model": None, "loaded_at": None}

def get_model():
    """モデルをキャッシュから取得（1時間ごとに再読み込み）"""
    now = datetime.now()
    if (_model_cache["model"] is None or
        _model_cache["loaded_at"] is None or
        now - _model_cache["loaded_at"] > timedelta(hours=1)):

        _model_cache["model"] = joblib.load("models/mood_predictor.pkl")
        _model_cache["loaded_at"] = now

    return _model_cache["model"]

@router.post("/predict/mood-risk")
async def predict_mood_risk(user_id: str, db: Session = Depends(get_db)):
    # 最近7日間のデータ取得
    logs = get_recent_logs(user_id, days=7, db=db)

    if len(logs) < 3:
        return {"error": "Insufficient data", "required_logs": 3}

    # 特徴量抽出
    features = extract_features(logs)

    # 予測
    model = get_model()
    risk_score = model.predict_proba([features])[0][1]

    return {
        "risk_level": "high" if risk_score > 0.7 else "low",
        "confidence": float(risk_score),
        "recommendation": get_recommendation(risk_score, logs)
    }

---

実装で直面した課題と解決策

課題1: バリデーションエラーが500エラーとして返される

問題の発見

テストで無効なスコア（11）を送信したところ、期待する422エラーではなく500エラーが返りました。

テストコード:

def test_create_action_log_invalid_score(client):
    """無効なスコア値のバリデーションテスト"""
    invalid_data = {
        "user_id": "test_user",
        "action_category": "運動",
        "mood_score": 15,  # 0-10の範囲外
        "energy_score": 5,
        "focus_score": 5
    }
    response = client.post("/api/v1/actions/", json=invalid_data)
    assert response.status_code == 422  # ❌ 実際は500が返る

原因分析

1. Pydanticスキーマに範囲制約がない:

   class ActionLogCreate(BaseModel):
       mood_score: int  # 制約なし！
       energy_score: int
       focus_score: int
   

2. DBのCHECK制約でエラー:

   # models/action_log.py
   mood_score = Column(
       Integer,
       CheckConstraint('mood_score >= 0 AND mood_score <= 10'),
       nullable=False
   )
   

3. 結果: Pydanticバリデーションをパスし、DB挿入時にIntegrityErrorが発生、500エラーとして返る

解決策

Pydanticスキーマにバリデーション制約を追加:

from pydantic import BaseModel, Field

class ActionLogCreate(BaseModel):
    user_id: str
    action_category: str
    mood_score: int = Field(..., ge=0, le=10, description="気分スコア (0-10)")
    energy_score: int = Field(..., ge=0, le=10, description="エネルギースコア (0-10)")
    focus_score: int = Field(..., ge=0, le=10, description="集中力スコア (0-10)")

修正後:

# テスト実行結果
response.status_code == 422  # ✅ 正しいステータスコード
response.json() == {
    "detail": [
        {
            "loc": ["body", "mood_score"],
            "msg": "ensure this value is less than or equal to 10",
            "type": "value_error.number.not_le"
        }
    ]
}

学び

• 多層防御の重要性: API層とDB層の両方でバリデーション
• 適切なエラーレスポンス: ユーザーに何が問題かを明確に伝える
• テストの価値: テストがなければ本番で発見されていた

---

課題2: テストでのタイムスタンプ競合

問題の発見

連続してログを作成するテストで、スコア差分計算が期待と異なりました。

テストの内容:

def test_mood_score_progression(client, sample_user_id):
    """気分スコアの変化を追跡するテスト"""
    mood_scores = [4, 5, 6, 7, 8]

    for i, mood in enumerate(mood_scores):
        response = client.post("/api/v1/actions/", json={
            "user_id": sample_user_id,
            "action_category": "運動",
            "mood_score": mood,
            "energy_score": 6,
            "focus_score": 6
        })

        if i > 0:
            data = response.json()
            expected_diff = mood - mood_scores[i-1]
            # ❌ 期待: 1, 実際: 2 などの不整合が発生
            assert data["mood_score_diff"] == expected_diff

原因分析

テスト実行が非常に高速なため、同じミリ秒内に複数のレコードが作成され、
order_by(timestamp.desc()).first()が意図しないレコードを取得していました。

元のコード:

# 前回のログを取得
last_log = db.query(ActionLog).filter(
    ActionLog.user_id == user_id
).order_by(ActionLog.timestamp.desc()).first()  # タイムスタンプで降順

問題:

• レコードA: timestamp = 2025-10-29 10:00:00.123
• レコードB: timestamp = 2025-10-29 10:00:00.123（同じミリ秒！）
• first()がどちらを返すか不定

解決策

IDベースの順序保証に変更:

# 前回のログを取得（IDで順序保証）
last_log = db.query(ActionLog).filter(
    ActionLog.user_id == user_id
).order_by(ActionLog.id.desc()).first()  # 自動インクリメントIDで降順

なぜIDが確実なのか:

• SQLiteの自動インクリメントIDは挿入順序を保証
• ミリ秒単位のタイムスタンプよりも確実
• データベースレベルでの順序保証

テスト結果:

# 修正後
test_mood_score_progression PASSED  # ✅ 成功

学び

• タイムスタンプだけに頼るのは危険: 特に高速処理では同時刻が発生
• 自動インクリメントIDによる順序保証: データベース機能を活用
• 実行速度の影響を考慮: 本番環境とテスト環境で挙動が変わることも

---

課題3: 機械学習モデルの統合とAPI設計

問題の発見

ML予測機能をどうAPIとして公開するか、パフォーマンスと保守性のバランスが課題でした。

検討した選択肢

選択肢A: リクエストごとにモデルをロード

@router.post("/predict")
async def predict(data: PredictionRequest):
    model = joblib.load("model.pkl")  # 毎回ディスクから読み込み
    prediction = model.predict(data.features)
    return {"prediction": prediction}

❌ 問題点:

• レイテンシが高い（~200ms）
• ディスクI/Oがボトルネック
• 同時リクエストでファイルロック競合

---

選択肢B: グローバル変数にキャッシュ

# アプリ起動時にロード
model = joblib.load("model.pkl")

@router.post("/predict")
async def predict(data: PredictionRequest):
    prediction = model.predict(data.features)
    return {"prediction": prediction}

❌ 問題点:

• モデル更新時にサーバー再起動が必要
• ダウンタイムが発生
• デプロイフローが複雑化

---

採用した解決策

LRUキャッシュ + 定期的な再読み込み:

from functools import lru_cache
from datetime import datetime, timedelta
import joblib

_model_cache = {"model": None, "loaded_at": None}

def get_model():
    """
    モデルをキャッシュから取得
    1時間ごとに自動的に再読み込み
    """
    now = datetime.now()

    # 初回ロードまたは1時間経過で再読み込み
    if (_model_cache["model"] is None or
        _model_cache["loaded_at"] is None or
        now - _model_cache["loaded_at"] > timedelta(hours=1)):

        print(f"[INFO] Loading ML model at {now}")
        _model_cache["model"] = joblib.load("models/mood_predictor.pkl")
        _model_cache["loaded_at"] = now

    return _model_cache["model"]

@router.post("/predict/mood-risk")
async def predict_mood_risk(data: PredictionRequest):
    model = get_model()  # キャッシュから取得
    prediction = model.predict(data.features)
    return {"risk_level": prediction}

メリット:

• ✅ 高速レスポンス: 2回目以降は~10msでレスポンス
• ✅ 自動更新: 1時間ごとに新しいモデルを読み込み
• ✅ ダウンタイムなし: サーバー再起動不要
• ✅ シンプル: 外部サービス（Redis等）が不要

パフォーマンス比較:

方式	初回	2回目以降	モデル更新
毎回ロード	200ms	200ms	即座
グローバル	0ms	10ms	再起動必要
キャッシュ	200ms	10ms	自動

学び

• トレードオフを理解する: 完璧な解決策はない、要件に最適な選択を
• パフォーマンス測定: 実測してボトルネックを特定
• 段階的な改善: まずシンプルに実装、必要に応じて高度化

---

テスト戦略と品質保証

テストピラミッドの実践

構成:

       /\
      /統合\     5テスト (18.5%)
     /------\    - エンドツーエンドワークフロー
    / API   \   12テスト (44.4%)
   /----------\  - エンドポイントテスト
  /ユニット   \  10テスト (37.0%)
 /--------------\ - モデル・ロジックテスト

達成した成果:

• 総テスト数: 27テスト（バックエンド）
• 成功率: 92.6%（25/27テスト成功）
• カバレッジ: actions.py 100%, models 100%
• 実行時間: 1.91秒（全テスト）

---

テストの具体例

1. ユニットテスト（モデル層）

境界値テスト:

import pytest
from app.models.action_log import ActionLog

@pytest.mark.parametrize("score", [0, 5, 10])
def test_mood_score_constraints(db_session, score):
    """境界値テスト: 0-10の範囲"""
    log = ActionLog(
        user_id="test_user",
        action_category="運動",
        mood_score=score,
        energy_score=5,
        focus_score=5
    )
    db_session.add(log)
    db_session.commit()

    assert log.mood_score == score
    assert log.id is not None  # 正常に保存された

デフォルト値テスト:

def test_action_log_defaults(db_session):
    """デフォルト値が正しく設定されるか"""
    log = ActionLog(
        user_id="test_user",
        action_category="交流",
        mood_score=7
    )
    db_session.add(log)
    db_session.commit()

    # タイムスタンプが自動設定される
    assert log.timestamp is not None
    assert log.created_at is not None

    # 差分はNoneで初期化
    assert log.mood_score_diff is None

---

2. APIテスト

正常系テスト:

def test_create_action_log_success(client, sample_user_id):
    """行動ログの正常作成"""
    data = {
        "user_id": sample_user_id,
        "action_category": "運動",
        "mood_score": 7,
        "energy_score": 6,
        "focus_score": 8,
        "detail": "30分のジョギング"
    }

    response = client.post("/api/v1/actions/", json=data)

    assert response.status_code == 200
    result = response.json()
    assert result["mood_score"] == 7
    assert result["action_category"] == "運動"
    assert "id" in result  # IDが生成されている

バリデーションエラーテスト:

def test_create_action_log_invalid_category(client, sample_user_id):
    """無効なカテゴリのバリデーション"""
    data = {
        "user_id": sample_user_id,
        "action_category": "無効なカテゴリ",
        "mood_score": 5,
        "energy_score": 5,
        "focus_score": 5
    }

    response = client.post("/api/v1/actions/", json=data)

    assert response.status_code == 422
    assert "action_category" in response.json()["detail"][0]["loc"]

フィルタリングテスト:

def test_get_action_logs_with_days_filter(client, sample_user_id):
    """日数フィルタが動作するか"""
    # 10日前のログを作成
    old_log = create_log(sample_user_id, days_ago=10)
    # 今日のログを作成
    new_log = create_log(sample_user_id, days_ago=0)

    # 7日間のログを取得
    response = client.get(f"/api/v1/actions/{sample_user_id}?days=7")

    result = response.json()
    assert len(result) == 1  # 今日のログのみ
    assert result[0]["id"] == new_log["id"]

---

3. 統合テスト

エンドツーエンドワークフロー:

def test_complete_action_log_workflow(client, sample_user_id):
    """
    完全なユーザーシナリオのテスト:
    1. ログ作成
    2. ログ一覧取得
    3. 統計取得
    4. データ整合性確認
    """
    # 1. 複数のログを作成
    logs_data = [
        {"mood_score": 5, "action_category": "運動"},
        {"mood_score": 7, "action_category": "交流"},
        {"mood_score": 6, "action_category": "趣味"},
    ]

    created_ids = []
    for data in logs_data:
        data["user_id"] = sample_user_id
        data["energy_score"] = 5
        data["focus_score"] = 5

        response = client.post("/api/v1/actions/", json=data)
        assert response.status_code == 200
        created_ids.append(response.json()["id"])

    # 2. ログ一覧取得
    response = client.get(f"/api/v1/actions/{sample_user_id}")
    assert response.status_code == 200
    logs = response.json()
    assert len(logs) == 3

    # 3. 統計取得
    response = client.get(f"/api/v1/actions/{sample_user_id}/stats")
    assert response.status_code == 200
    stats = response.json()

    # 4. データ整合性確認
    assert stats["total_logs"] == 3
    assert stats["avg_mood"] == (5 + 7 + 6) / 3  # 6.0

マルチユーザー分離テスト:

def test_multi_user_isolation(client):
    """異なるユーザーのデータが分離されているか"""
    user1_id = "user_001"
    user2_id = "user_002"

    # User1のログ作成
    create_log_for_user(client, user1_id, mood_score=8)

    # User2のログ作成
    create_log_for_user(client, user2_id, mood_score=3)

    # User1は自分のデータのみ取得
    response = client.get(f"/api/v1/actions/{user1_id}")
    logs = response.json()
    assert len(logs) == 1
    assert logs[0]["mood_score"] == 8

    # User2は自分のデータのみ取得
    response = client.get(f"/api/v1/actions/{user2_id}")
    logs = response.json()
    assert len(logs) == 1
    assert logs[0]["mood_score"] == 3

---

テストフィクスチャ

conftest.py で共通設定:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from main import app
from app.core.database import Base, get_db

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

@pytest.fixture(scope="function")
def db_engine():
    """テスト用DBエンジンを作成"""
    engine = create_engine(
        SQLALCHEMY_DATABASE_URL,
        connect_args={"check_same_thread": False}
    )
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def db_session(db_engine):
    """テスト用DBセッションを作成"""
    SessionLocal = sessionmaker(bind=db_engine)
    session = SessionLocal()
    yield session
    session.close()

@pytest.fixture(scope="function")
def client(db_session):
    """テスト用クライアントを作成"""
    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

@pytest.fixture
def sample_user_id():
    """テスト用ユーザーID"""
    return "test_user_12345"

---

CI/CDでの自動テスト

GitHub Actions設定:

# .github/workflows/test.yml
name: Tests

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  backend-test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.9'

      - name: Install dependencies
        run: |
          cd backend
          pip install -r requirements.txt

      - name: Run pytest with coverage
        run: |
          cd backend
          pytest --cov=app --cov-report=xml --cov-report=term-missing

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v3
        with:
          file: ./backend/coverage.xml
          flags: backend

  frontend-test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Node.js
        uses: actions/setup-node@v3
        with:
          node-version: '18'

      - name: Install dependencies
        run: |
          cd frontend
          npm install

      - name: Run Jest tests
        run: |
          cd frontend
          npm test -- --coverage

---

カバレッジ詳細

ファイル	ステートメント数	カバー	カバレッジ	未カバー行
app/api/actions.py	79	79	100% ✅	-
app/models/action_log.py	17	17	100% ✅	-
app/models/prediction.py	11	11	100% ✅	-
app/core/config.py	36	34	94.44% ✅	34-35
app/core/database.py	30	12	40.00% ⚠️	17-21, 31-60
app/api/predictions.py	117	27	23.08% ⚠️	多数
合計	293	183	62.46%	-

改善の優先順位:

1. 🔴 predictions.py のテスト追加（現在23% → 目標80%）
2. 🟡 database.py の接続エラーハンドリングテスト
3. 🟢 エッジケースのカバレッジ向上

---

学び

• テストは投資: 2件のバグを本番前に発見
• カバレッジ100%が目標ではない: 重要な部分を確実にテスト
• テストピラミッド: ユニット > API > 統合の比率を意識
• CI/CD統合: プッシュのたびに自動テストで品質担保

---

パフォーマンス最適化

フロントエンドの最適化

1. Server Componentsの活用

// app/dashboard/page.tsx (Server Component)
export default async function DashboardPage() {
  // サーバーサイドでデータ取得
  // クライアント側のJSバンドルに含まれない
  const logs = await getActionLogs(userId)

  return (
    <div>
      <h1>ダッシュボード</h1>
      <Dashboard initialData={logs} />
    </div>
  )
}

メリット:

• クライアント側のJSバンドルサイズ削減
• 初期表示が高速
• SEO対策にも有効

---

2. 動的インポートでバンドルサイズ削減

import dynamic from 'next/dynamic'

// グラフライブラリを動的ロード（重い依存関係）
const Chart = dynamic(() => import('@/components/MoodChart'), {
  ssr: false,  // サーバーサイドレンダリング無効
  loading: () => <Skeleton className="h-64 w-full" />
})

export default function Dashboard({ data }) {
  return (
    <div>
      <Chart data={data} />
    </div>
  )
}

効果:

• 初期バンドルサイズ: 320KB → 180KB（44%削減）
• 必要なときだけロード

---

3. 画像最適化

import Image from 'next/image'

export function AppIcon() {
  return (
    <Image
      src="/icon-192.png"
      width={192}
      height={192}
      alt="App Icon"
      priority  // LCP（Largest Contentful Paint）向上
      placeholder="blur"
      blurDataURL="data:image/..."
    />
  )
}

---

4. データフェッチング最適化

// SWRでクライアント側キャッシング
import useSWR from 'swr'

function useActionLogs(userId: string) {
  const { data, error, mutate } = useSWR(
    `/api/v1/actions/${userId}`,
    fetcher,
    {
      revalidateOnFocus: false,  // フォーカス時の再検証無効
      dedupingInterval: 5000,    // 5秒以内の重複リクエスト防止
    }
  )

  return { logs: data, isLoading: !error && !data, mutate }
}

---

Lighthouse Score

結果:

• Performance: 95点
• Accessibility: 100点
• Best Practices: 100点
• SEO: 100点

主要指標:

• FCP（First Contentful Paint）: 1.2秒
• LCP（Largest Contentful Paint）: 1.8秒
• TTI（Time to Interactive）: 2.3秒
• CLS（Cumulative Layout Shift）: 0.05

---

バックエンドの最適化

1. データベースクエリ最適化

N+1問題の回避:

# ❌ N+1問題が発生
logs = db.query(ActionLog).filter(ActionLog.user_id == user_id).all()
for log in logs:
    user = log.user  # 各ログごとにSELECT文が発行される

# ✅ Eager loading
from sqlalchemy.orm import joinedload

logs = db.query(ActionLog).filter(
    ActionLog.user_id == user_id
).options(
    joinedload(ActionLog.user)  # JOINで一度に取得
).all()

インデックスの追加:

# models/action_log.py
class ActionLog(Base):
    __tablename__ = "action_logs"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(String, index=True)  # 検索頻度が高いのでインデックス
    timestamp = Column(DateTime, index=True)  # 日付範囲検索でインデックス

---

2. 非同期処理

from fastapi import BackgroundTasks

@router.post("/actions/")
async def create_action(
    action: ActionLogCreate,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    # 同期処理（即座に返す）
    db_action = ActionLog(**action.dict())
    db.add(db_action)
    db.commit()
    db.refresh(db_action)

    # 重い処理はバックグラウンドで実行
    background_tasks.add_task(
        update_ml_model_async,
        user_id=action.user_id
    )

    return db_action

効果:

• レスポンスタイム: 150ms → 50ms（67%改善）
• ユーザー体験の向上

---

3. キャッシング戦略

from functools import lru_cache

@lru_cache(maxsize=100)
def get_user_stats(user_id: str, days: int = 7) -> dict:
    """
    ユーザー統計をキャッシュ
    同じパラメータでの再計算を防ぐ
    """
    logs = get_logs(user_id, days)
    return calculate_stats(logs)

# キャッシュクリア（新規ログ作成時）
@router.post("/actions/")
async def create_action(action: ActionLogCreate):
    # ログ作成
    db_action = create_log(action)

    # キャッシュクリア
    get_user_stats.cache_clear()

    return db_action

---

4. データベース接続プール

# app/core/database.py
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    poolclass=QueuePool,
    pool_size=5,          # 常時5接続を維持
    max_overflow=10,      # 最大15接続まで許可
    pool_timeout=30,      # タイムアウト30秒
    pool_recycle=3600,    # 1時間ごとに接続をリサイクル
)

---

パフォーマンス測定結果

API レスポンスタイム（平均）:

エンドポイント	最適化前	最適化後	改善率
POST /actions/	120ms	45ms	62% ⬇️
GET /actions/{user_id}	85ms	30ms	65% ⬇️
GET /stats/{user_id}	200ms	60ms	70% ⬇️
POST /predict/mood-risk	250ms	15ms	94% ⬇️

---

デプロイとインフラ

デプロイ構成

全体アーキテクチャ:

[ユーザー]
    ↓
[Vercel CDN] ← Next.js (Frontend)
    ↓ REST API
[Render] ← FastAPI (Backend)
    ↓
[SQLite/PostgreSQL]

---

フロントエンド（Vercel）

デプロイ設定:

// vercel.json
{
  "buildCommand": "npm run build",
  "outputDirectory": ".next",
  "devCommand": "npm run dev",
  "installCommand": "npm install",
  "framework": "nextjs",
  "regions": ["hnd1"],  // 東京リージョン
  "env": {
    "NEXT_PUBLIC_API_URL": "@api-url",
    "NEXT_PUBLIC_FIREBASE_API_KEY": "@firebase-api-key"
  }
}

自動デプロイ:

• mainブランチへのプッシュで本番デプロイ
• PRごとにプレビュー環境自動生成
• ロールバック機能

---

バックエンド（Render）

render.yaml:

services:
  - type: web
    name: action-therapy-api
    env: python
    buildCommand: "pip install -r requirements.txt"
    startCommand: "uvicorn main:app --host 0.0.0.0 --port $PORT"
    envVars:
      - key: SECRET_KEY
        generateValue: true
      - key: DATABASE_URL
        fromDatabase:
          name: action-therapy-db
          property: connectionString
    healthCheckPath: /health

databases:
  - name: action-therapy-db
    databaseName: action_therapy
    user: app_user

---

環境変数管理

3箇所で設定が必要:

1. ローカル開発（.env.local）

# Frontend
NEXT_PUBLIC_API_URL=http://localhost:8000/api/v1
NEXT_PUBLIC_FIREBASE_API_KEY=...
NEXT_PUBLIC_FIREBASE_AUTH_DOMAIN=...

# Backend
SECRET_KEY=dev-secret-key-change-in-production
DATABASE_URL=sqlite:///./action_therapy.db

2. Vercel（プロダクション）

# Vercel Dashboard → Settings → Environment Variables
NEXT_PUBLIC_API_URL=https://api.example.com/api/v1
NEXT_PUBLIC_FIREBASE_API_KEY=...

3. GitHub Actions（CI/CD）

# .github/workflows/deploy.yml
env:
  NEXT_PUBLIC_API_URL: ${{ secrets.NEXT_PUBLIC_API_URL }}
  NEXT_PUBLIC_FIREBASE_API_KEY: ${{ secrets.NEXT_PUBLIC_FIREBASE_API_KEY }}

---

CI/CDパイプライン

デプロイフロー:

GitHub Actions workflow:

name: Deploy

on:
  push:
    branches: [main]

jobs:
  test-and-deploy:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Run Backend Tests
        run: |
          cd backend
          pip install -r requirements.txt
          pytest

      - name: Run Frontend Tests
        run: |
          cd frontend
          npm install
          npm test

      - name: Deploy to Vercel
        if: success()
        run: |
          npm install -g vercel
          vercel --prod --token=${{ secrets.VERCEL_TOKEN }}

      - name: Deploy to Render
        if: success()
        run: |
          curl -X POST ${{ secrets.RENDER_DEPLOY_HOOK }}

---

セキュリティ対策

1. 環境変数の保護

# backend/app/core/config.py
import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = os.getenv("SECRET_KEY")

    class Config:
        env_file = ".env"
        case_sensitive = True
        extra = "ignore"

# 環境変数が設定されていない場合はエラー
if not Settings().SECRET_KEY or Settings().SECRET_KEY == "dev-secret-key":
    raise ValueError("SECRET_KEY must be set in production")

2. CORS設定

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://your-frontend.vercel.app",
        "http://localhost:3000"  # 開発環境のみ
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

3. レート制限

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/api/v1/actions/")
@limiter.limit("10/minute")  # 1分間に10リクエストまで
async def create_action(request: Request, action: ActionLogCreate):
    # ...

---

モニタリング

ヘルスチェックエンドポイント:

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0"
    }

ログ収集:

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

@app.post("/actions/")
async def create_action(action: ActionLogCreate):
    logger.info(f"Creating action log for user {action.user_id}")
    # ...

---

今後の展望

Phase 2: ML機能の強化

• LSTMによる時系列予測: より精度の高い気分予測
• 協調フィルタリング: 類似ユーザーからの推薦
• A/Bテスト: 推薦アルゴリズムの効果測定
• 特徴量エンジニアリング: 天気データ、曜日、時間帯の統合

技術選定:

• PyTorch/TensorFlow for LSTM
• MLflowでモデルバージョン管理
• Kubeflowでパイプライン自動化

---

Phase 3: スケーラビリティ

• SQLite → PostgreSQL移行: 本格的な運用に向けて
• Redis導入: セッション管理、キャッシング層
• ML予測サービス分離: マイクロサービス化
• 負荷分散: 複数インスタンスでのスケーリング

アーキテクチャ（将来）:

[Next.js] → [API Gateway]
              ↓
         [FastAPI (CRUD)]
              ↓
         [PostgreSQL]

         [ML Service (Python)]
              ↓
         [Model Registry]

---

Phase 4: ユーザー体験

• リアルタイム通知: WebSocketで即座にアラート
• オフライン対応強化: Service Workerでオフライン編集
• ダークモード: UI/UXの向上
• 多言語対応: 国際化（i18n）
• アクセシビリティ向上: WCAG 2.1 AAA準拠

---

技術的負債の返済

優先度高:

• 予測APIのテストカバレッジ向上（現在23% → 目標80%）
• E2Eテスト導入（Playwright）
• パフォーマンステスト自動化（Locust）

優先度中:

• TypeScript strict mode有効化
• ESLint/Prettierルールの厳格化
• Docker Composeでローカル環境統一

優先度低:

• OpenAPIスキーマの自動生成強化
• Storybook導入（コンポーネントカタログ）

---

まとめ

学んだこと

1. 技術選定の重要性

FastAPI + Next.jsの組み合わせは、フルスタック開発で優れた生産性を発揮しました。
特にPythonでMLと統合できる点が大きなメリットでした。

2. テストファーストの開発

テストを先に書くことで、バリデーションエラーやタイムスタンプ競合など、
重要なバグを本番前に発見できました。テストは「コスト」ではなく「投資」です。

3. パフォーマンスと保守性のバランス

MLモデルのキャッシング戦略では、複数の選択肢を比較検討し、
トレードオフを理解した上で最適な解決策を選びました。

---

このプロジェクトで習得したスキル

• ✅ フルスタック開発（TypeScript/Python）
• ✅ RESTful API設計とOpenAPI仕様書作成
• ✅ 機械学習の実装と統合（scikit-learn）
• ✅ テスト駆動開発（TDD）、92.6%の成功率達成
• ✅ CI/CD構築（GitHub Actions）
• ✅ パフォーマンス最適化（67%のレスポンス改善）
• ✅ セキュアな環境変数管理
• ✅ PWA開発とLighthouse最適化（95点達成）

---

📞 連絡先

現在、機械学習・バックエンド・フルスタックエンジニアとして就職・転職活動中です。カジュアル面談も歓迎です。

📧 Email: [e-mail]
🐙 GitHub: https://github.com/Datarchpy
📁 本プロジェクト: https://github.com/Datarchpy/action-activation-therapy

ご興味を持っていただけましたら、お気軽にご連絡ください。