Chapter 10: 統合実践プロジェクト 🏗️


📚 目次に戻る: 📖 学習ガイド
⬅️ 前の章: Chapter 9: アーキテクチャ選択演習
➡️ 完了: 🎓 学習修了・次のステップ
🎯 学習フェーズ: Part IV - 実践・応用編(統合実践)
🎯 学習レベル: 🌱 基礎 | 🚀 応用 | 💪 発展
⏱️ 推定学習時間: 8-12時間
📝 難易度: 上級(全章の知識統合が必要) —

🏗️ 総合建設プロジェクト
├── 🏠 住宅棟(個人学習用):木造建築でシンプル
├── 🏢 オフィス棟(コース管理):プレハブ工法で柔軟
├── 🏦 本社ビル(管理機能):鉄筋コンクリートで堅牢
└── 🔌 共通インフラ:電気・水道・通信を全棟で共有

💾 EduConnect統合プラットフォーム
├── 🏠 学習者ポータル:Python Fletでデスクトップアプリ
├── 🏢 コース管理:Deno Edge Functionsで動画配信
├── 🏦 管理ダッシュボード:FastAPIで高度な分析
└── 🔌 Supabaseインフラ:データベース・認証・ストレージ

🎯 この章で学ぶこと

これまでの章では、3つの建築工法を別々に学習してきました。この章では、それらを1つの大規模複合施設として統合し、実際の商業プロジェクトレベルの開発を体験します。

🌱 初心者が陥りがちな問題

# ❌ よくある初心者の統合失敗パターン
とりあえず全部APIサーバーで作ろう  # → 過剰設計
後で統合すればいいでしょ        # → 技術的負債
セキュリティは最後に考える      # → 設計やり直し
# → 結果:プロジェクト炎上・予算超過・品質問題

✅ この章で身につく統合力

# ✅ プロの統合設計アプローチ
適材適所_設計 = {
    "学習者ポータル": "Client-side",    # 個人用途→木造住宅
    "コース管理": "Edge Functions",     # 外部連携→プレハブ工法  
    "管理機能": "API Server",          # 企業用途→鉄筋コンクリート
    "共通インフラ": "Supabase"          # 全棟共通設備
}
# → 結果:適切なコスト・最高の性能・保守しやすい設計

📚 学習進度別ガイド

レベル 対象者 学習内容 所要時間
🌱 基礎 各パターンは理解している方 統合設計・データ共有・認証統一 8-12時間
🚀 応用 統合の基本ができる方 CI/CD・監視・スケーリング・障害対応 12-18時間
💪 発展 本格運用を目指す方 エンタープライズ運用・法的対応・事業継続 18-25時間

🎯 最終成果物

成果物 内容 ビジネス価値
本格的なWebアプリ 3パターン統合システム 実際の企業で使える品質
運用環境一式 CI/CD・監視・バックアップ 24時間365日運用可能
技術ドキュメント 設計書・運用手順・障害対応 チーム開発・引き継ぎ対応
ポートフォリオ GitHub・デモサイト・技術説明 転職・案件獲得に使える

10.1 プロジェクト要件定義(複合施設建設計画)

実際の教育会社が「全社のデジタル学習プラットフォームを作りたい」と依頼してきた想定で、要件定義から運用開始まで一貫して体験しましょう。

🏢 クライアント企業プロフィール

企業名: 東京教育アカデミー株式会社
事業内容: 大学受験予備校・オンライン教育・企業研修
規模: 従業員200名、年間受講生15,000人
予算: 年間IT予算2,000万円
期間: 6ヶ月での本格運用開始

📋 ビジネス要件(何を作るか)

🎯 解決したい問題

| 現在の問題 | 希望する解決策 | 期待効果 | |:———-|:————-|:———| | 紙ベースの学習管理 | デジタル学習ポータル | 効率3倍向上 | | 講師の手動作業多数 | 自動化システム | 人件費30%削減 | | 経営データの不足 | リアルタイム分析 | 意思決定速度向上 | | 生徒の離脱率高い | 個別最適化学習 | 継続率20%向上 |

🎯 主要機能要件

💻 学習者ポータル(一般ユーザー向け)

  • オフライン学習対応(電車内でも使える)
  • 進捗の自動同期
  • リアルタイム質問・チャット
  • 親御さんへの進捗レポート

🏢 コース管理システム(講師・教材管理)

  • 動画コンテンツ配信(適応的画質)
  • 自動採点システム
  • 証明書・修了証の自動生成
  • 外部教材システム連携

📊 管理ダッシュボード(経営陣・管理者向け)

  • 学習進捗の統計分析
  • 売上・コスト分析
  • 法的要件対応(個人情報保護等)
  • マルチテナント対応(支校別管理)

🛡️ 非機能要件(どの程度の品質で)

項目 要件 理由・背景
可用性 99.9%以上 「試験前に使えない」は絶対NG
レスポンス時間 500ms以下 学習集中を妨げない
同時接続数 10,000人 試験前の集中アクセス対応
データ保持 5年間 卒業後も成績証明書発行
セキュリティ 高レベル 個人情報・成績情報の保護
スケーラビリティ 年20%成長 事業拡大に対応

🔰 初心者向け解説:要件定義の重要性

実際のビジネスでは、「なんとなく便利そう」ではシステムは作れません。レストランを開業するとき:

レストラン開業 システム開発 失敗する例
「どんな客層?」 「どんなユーザー?」 「みんな使うでしょ」
「何人席?」 「何人同時利用?」 「そんなの後で考える」
「予算は?」 「開発・運用予算は?」 「安くできるでしょ」
「いつオープン?」 「いつリリース?」 「できるだけ早く」

このように曖昧な要件だと、途中で「話が違う」となり、プロジェクトが炎上します。

🏗️ アーキテクチャ戦略(複合施設建設計画)

複合施設を建設するとき、用途に応じて最適な建築工法を選択し、共通インフラで効率的に運営します。

📊 建物ごとの工法選択理由

建物 利用者 使用頻度 選択工法 選択理由
🏠 住宅棟 学習者 毎日使用 木造(Client-side) 個人利用・オフライン対応・高速起動
🏢 オフィス棟 講師 業務時間 プレハブ(Edge Functions) 外部連携・柔軟性・自動スケール
🏦 本社ビル 経営陣 分析時 鉄筋(API Server) 複雑分析・高セキュリティ・大容量

🔌 共通インフラ(ライフライン)

インフラ 実際の施設 Supabaseサービス 役割
電力供給 電力会社 Supabase Auth 全建物の認証・ログイン
上下水道 水道局 PostgreSQL Database 全データの保存・管理
通信網 電話・インターネット Supabase Realtime リアルタイム通信
廃棄物処理 清掃業者 Supabase Storage ファイル・画像・動画保存

🔄 システム連携フロー

📱 学習者の1日の流れ
👨‍🎓 朝: デスクトップアプリで学習開始 → [木造住宅]
📚 昼: 動画教材を視聴 → [プレハブ棟の配信機能]
📊 夜: 進捗確認、親への報告 → [全建物のデータ統合]

👨‍🏫 講師の1日の流れ  
📝 朝: 課題の自動採点結果確認 → [プレハブ棟の採点機能]
🎥 昼: 新しい動画教材をアップロード → [共通ストレージ]
📈 夕: 学習状況分析レポート確認 → [本社ビルの分析機能]

👔 経営陣の月次業務
📊 月初: 全体の学習進捗分析 → [本社ビルの高度分析]
💰 月中: 売上・コスト分析 → [本社ビルの経営ダッシュボード]
📋 月末: 支校別実績比較 → [マルチテナント機能]

🔰 初心者向け解説

このように適材適所で技術を選択することで:

コスト最適化: 必要最小限の技術で最大効果
性能最適化: 各用途に最適な工法で高性能
保守性: 複雑さを適切に分散して管理しやすく
スケーラビリティ: 各部分を独立してスケール可能


10.2 Phase 1: プロジェクト基盤構築(建設現場の準備)

複合施設建設と同じように、まず建設現場を整備してから、各建物の工事を開始します。

🎯 Step 1: 統合開発環境セットアップ(建設現場の整備)

建設現場を準備するとき、各建物用の作業エリア共通設備エリアを分けて整備します:

📁 プロジェクト構造(建設現場のレイアウト)

# 🏗️ EduConnect建設現場の全体レイアウト
educonnect/
├── 🏠 frontend/              # 住宅棟建設エリア(学習者ポータル)
│   ├── src/                  # 設計図・部材
│   ├── requirements.txt      # 必要な工具・材料リスト
│   ├── Dockerfile           # 建設手順書
│   └── tests/               # 品質検査手順
├── 🏢 edge-functions/        # オフィス棟建設エリア(コース管理)
│   ├── course-management/    # コース管理部屋
│   ├── video-streaming/      # 動画配信部屋
│   ├── certificate-generator/ # 証明書発行部屋
│   └── tests/               # 品質検査手順
├── 🏦 backend/              # 本社ビル建設エリア(管理ダッシュボード)
│   ├── app/                 # アプリケーション本体
│   ├── requirements.txt     # 必要なライブラリ
│   ├── Dockerfile          # 建設手順書
│   └── tests/              # 品質検査手順
├── 🔌 infrastructure/       # 共通設備エリア(電気・水道・通信)
│   ├── docker-compose.yml  # 全体設備配置図
│   ├── supabase/           # データベース設備
│   ├── monitoring/         # 監視システム
│   └── backup/             # バックアップ設備
├── 📚 docs/                 # 設計書・取扱説明書
│   ├── architecture.md     # 全体設計書
│   ├── api-docs/           # API設計書
│   └── deployment.md       # 運用手順書
└── 🛠️ scripts/             # 建設・運用用スクリプト
    ├── setup.sh            # 初期セットアップ
    ├── deploy.sh           # デプロイメント
    └── backup.sh           # バックアップ実行

🔰 初心者向け解説

このディレクトリ構造は、建設現場の区画分けと同じ考え方です:

フォルダ 建設現場の例 役割 理由
frontend/ 住宅建設エリア 個人向けアプリ 他の工事と干渉しない
edge-functions/ プレハブ組立エリア サーバー機能 モジュール化して管理
backend/ 本社ビル建設エリア 管理システム 高度な作業で専用区画
infrastructure/ 設備エリア 共通インフラ 全建物で共有するもの
docs/ 現場事務所 設計書・手順書 情報を一元管理

🎯 Step 2: 共通インフラ設定(ライフライン整備)

建物を建てる前に、電気・水道・通信などのライフラインを整備します:

# 🔌 docker-compose.yml - 複合施設のライフライン設計図
version: '3.8'

services:
  # 🏭 Supabase基盤インフラ(発電所・浄水場・通信局を統合した巨大施設)
  supabase:
    image: supabase/supabase:latest
    ports:
      - "54321:54321"  # 🎨 Studio(管理用コントロールルーム)
      - "54322:54322"  # 🔌 PostgREST(データアクセス専用回線)
      - "54323:54323"  # 🔐 Auth(セキュリティゲート)
      - "54324:54324"  # 📡 Realtime(ライブ通信アンテナ)
    environment:
      POSTGRES_PASSWORD: postgres                    # 🔑 管理者パスワード
      ANON_KEY: ${SUPABASE_ANON_KEY}                # 🎫 一般利用者用アクセスキー
      SERVICE_ROLE_KEY: ${SUPABASE_SERVICE_ROLE_KEY} # 🔐 システム管理者用マスターキー
    volumes:
      - supabase_data:/var/lib/supabase             # 💾 永続データ保存庫
    networks:
      - educonnect_network

  # 🏠 学習者ポータル(住宅棟)
  learner-portal:
    build:
      context: ./frontend      # 🏗️ 住宅棟の設計図がある場所
      dockerfile: Dockerfile   # 🔨 建設手順書
    ports:
      - "8080:8080"           # 🚪 住宅棟の入口(ポート8080番)
    environment:
      SUPABASE_URL: http://supabase:54321           # 🔌 共通インフラへの接続先
      SUPABASE_ANON_KEY: ${SUPABASE_ANON_KEY}      # 🎫 住民用アクセスパス
    depends_on:
      - supabase              # 🏭 インフラが準備できてから建設開始
    networks:
      - educonnect_network

  # 🏦 管理ダッシュボード(本社ビル)
  admin-dashboard:
    build:
      context: ./backend       # 🏗️ 本社ビルの設計図がある場所
      dockerfile: Dockerfile   # 🔨 建設手順書
    ports:
      - "8000:8000"           # 🚪 本社ビルの入口(ポート8000番)
    environment:
      # 🔌 データベース直結専用線(本社ビル特権)
      DATABASE_URL: postgresql://postgres:postgres@supabase:5432/postgres
      SUPABASE_URL: http://supabase:54321           # 🔌 共通インフラへの接続
      SUPABASE_SERVICE_KEY: ${SUPABASE_SERVICE_ROLE_KEY}  # 🔐 管理者専用マスターキー
    depends_on:
      - supabase              # 🏭 インフラが準備できてから
      - redis                 # 📦 高速キャッシュシステムが準備できてから
    networks:
      - educonnect_network

  # 📦 Redis(高速キャッシュ倉庫)
  redis:
    image: redis:7.2-alpine   # 🏪 軽量・高性能な倉庫
    ports:
      - "6379:6379"           # 🚪 倉庫の出入口
    volumes:
      - redis_data:/data      # 💾 キャッシュデータ保存場所
    networks:
      - educonnect_network

  # 📊 監視システム(警備・監視センター)
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"           # 🖥️ 監視センターのダッシュボード
    volumes:
      # 📋 監視項目設定書
      - ./infrastructure/monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    networks:
      - educonnect_network

  # 📈 可視化ダッシュボード(総合管制室)
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"           # 🖥️ 管制室の大型ディスプレイ
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin  # 🔑 管制室入室パスワード
    volumes:
      - grafana_data:/var/lib/grafana    # 💾 ダッシュボード設定保存
    networks:
      - educonnect_network

# 💾 データ永続化ボリューム(地下貯蔵庫)
volumes:
  supabase_data:              # 🏭 メインデータベース貯蔵庫
  redis_data:                 # 📦 キャッシュデータ貯蔵庫
  grafana_data:               # 📊 監視データ貯蔵庫

# 🌐 ネットワーク(建物間通信網)
networks:
  educonnect_network:         # 🔌 全建物を結ぶ専用通信網
    driver: bridge            # 🌉 ブリッジ型ネットワーク(安定・高速)

🔰 初心者向け解説

このDocker Composeファイルは、複合施設の設備配置図と同じです:

設備 実際の施設 役割 ポート
supabase 発電所・浄水場 全建物の基盤サービス 54321-54324
learner-portal 住宅棟 学習者向けアプリ 8080
admin-dashboard 本社ビル 管理者向けシステム 8000
redis 高速倉庫 データキャッシュ 6379
prometheus 監視センター システム監視 9090
grafana 管制室 監視データ可視化 3000

🎯 Step 3: 環境変数設定(セキュリティキー管理)

各建物の鍵や暗証番号を安全に管理するため、環境変数ファイルを作成します:

# 📄 .env - セキュリティキー管理ファイル
# ⚠️ このファイルは絶対にGitにコミットしないこと!

# 🔐 Supabase認証キー(建物の入場パス)
SUPABASE_URL=http://localhost:54321
SUPABASE_ANON_KEY=your_anon_key_here        # 🎫 一般利用者パス
SUPABASE_SERVICE_ROLE_KEY=your_service_key_here  # 🔑 管理者マスターキー

# 🗄️ データベース接続情報(金庫の暗証番号)
DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres

# 🔒 セキュリティ設定
JWT_SECRET=your_jwt_secret_here             # 🔐 トークン暗号化キー
ENCRYPTION_KEY=your_encryption_key_here     # 🔐 データ暗号化キー

# 🌍 本番環境設定
ENVIRONMENT=development                     # 環境(development/staging/production)
DEBUG=true                                 # デバッグモード(本番では必ずfalse)

# 📧 通知設定
SMTP_SERVER=smtp.gmail.com                # メール送信サーバー
SMTP_USERNAME=your_email@gmail.com        # メール送信用アカウント
SMTP_PASSWORD=your_app_password           # アプリパスワード

# 📊 監視・ログ設定
LOG_LEVEL=INFO                            # ログレベル
MONITORING_ENABLED=true                   # 監視機能の有効化

supabase_data:


### 1.2 統合データベース設計

```sql
-- supabase/migrations/001_initial_schema.sql

-- 共通スキーマ
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- 組織・テナント管理
CREATE TABLE organizations (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    name TEXT NOT NULL,
    domain TEXT UNIQUE NOT NULL,
    plan_type TEXT NOT NULL DEFAULT 'basic',
    settings JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- ユーザープロファイル
CREATE TABLE user_profiles (
    id UUID PRIMARY KEY REFERENCES auth.users(id) ON DELETE CASCADE,
    organization_id UUID REFERENCES organizations(id),
    email TEXT NOT NULL,
    full_name TEXT,
    role TEXT NOT NULL DEFAULT 'student',
    avatar_url TEXT,
    preferences JSONB DEFAULT '{}',
    last_active_at TIMESTAMPTZ DEFAULT NOW(),
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- コース管理
CREATE TABLE courses (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    organization_id UUID REFERENCES organizations(id),
    title TEXT NOT NULL,
    description TEXT,
    instructor_id UUID REFERENCES user_profiles(id),
    category TEXT NOT NULL,
    difficulty_level INTEGER DEFAULT 1,
    estimated_hours INTEGER,
    thumbnail_url TEXT,
    status TEXT DEFAULT 'draft',
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- 学習進捗
CREATE TABLE learning_progress (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    user_id UUID REFERENCES user_profiles(id),
    course_id UUID REFERENCES courses(id),
    lesson_id UUID,
    completion_percentage DECIMAL(5,2) DEFAULT 0,
    time_spent_minutes INTEGER DEFAULT 0,
    last_accessed_at TIMESTAMPTZ DEFAULT NOW(),
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(user_id, course_id, lesson_id)
);

-- 分析・レポート用ビュー
CREATE VIEW course_analytics AS
SELECT 
    c.id as course_id,
    c.title,
    c.organization_id,
    COUNT(DISTINCT lp.user_id) as enrolled_students,
    AVG(lp.completion_percentage) as avg_completion,
    SUM(lp.time_spent_minutes) as total_time_spent,
    COUNT(CASE WHEN lp.completion_percentage = 100 THEN 1 END) as completed_count
FROM courses c
LEFT JOIN learning_progress lp ON c.id = lp.course_id
GROUP BY c.id, c.title, c.organization_id;

-- RLSポリシー設定
ALTER TABLE organizations ENABLE ROW LEVEL SECURITY;
ALTER TABLE user_profiles ENABLE ROW LEVEL SECURITY;
ALTER TABLE courses ENABLE ROW LEVEL SECURITY;
ALTER TABLE learning_progress ENABLE ROW LEVEL SECURITY;

-- 組織レベルのデータ分離
CREATE POLICY "org_isolation_organizations" ON organizations
    FOR ALL USING (
        id IN (
            SELECT organization_id 
            FROM user_profiles 
            WHERE id = auth.uid()
        )
    );

CREATE POLICY "org_isolation_users" ON user_profiles
    FOR ALL USING (
        organization_id IN (
            SELECT organization_id 
            FROM user_profiles 
            WHERE id = auth.uid()
        )
    );

CREATE POLICY "org_isolation_courses" ON courses
    FOR ALL USING (
        organization_id IN (
            SELECT organization_id 
            FROM user_profiles 
            WHERE id = auth.uid()
        )
    );

CREATE POLICY "progress_owner_only" ON learning_progress
    FOR ALL USING (
        user_id = auth.uid() OR
        EXISTS (
            SELECT 1 FROM user_profiles 
            WHERE id = auth.uid() 
            AND role IN ('admin', 'instructor')
            AND organization_id = (
                SELECT organization_id FROM user_profiles WHERE id = learning_progress.user_id
            )
        )
    );

1.3 統合認証システム

# backend/app/core/auth.py
from typing import Optional, List
from fastapi import HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from supabase import create_client
import jwt
from pydantic import BaseModel

class UserContext(BaseModel):
    id: str
    email: str
    organization_id: str
    role: str
    permissions: List[str]

class AuthManager:
    def __init__(self, supabase_url: str, supabase_key: str):
        self.supabase = create_client(supabase_url, supabase_key)
        self.security = HTTPBearer()
    
    async def get_current_user(
        self, 
        credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer())
    ) -> UserContext:
        """統合認証:全パターンで共通利用"""
        
        try:
            # JWTトークン検証
            token = credentials.credentials
            user_response = await self.supabase.auth.get_user(token)
            
            if user_response.user is None:
                raise HTTPException(status_code=401, detail="Invalid token")
            
            # ユーザープロファイル取得
            profile_response = await self.supabase.table('user_profiles')\
                .select('*')\
                .eq('id', user_response.user.id)\
                .single()\
                .execute()
            
            if not profile_response.data:
                raise HTTPException(status_code=404, detail="User profile not found")
            
            profile = profile_response.data
            
            # ロール・権限取得
            permissions = await self._get_user_permissions(
                profile['role'], 
                profile['organization_id']
            )
            
            return UserContext(
                id=profile['id'],
                email=profile['email'],
                organization_id=profile['organization_id'],
                role=profile['role'],
                permissions=permissions
            )
            
        except Exception as e:
            raise HTTPException(status_code=401, detail=f"Authentication failed: {str(e)}")
    
    async def _get_user_permissions(self, role: str, org_id: str) -> List[str]:
        """ロールベース権限管理"""
        
        role_permissions = {
            'student': [
                'courses:read',
                'progress:read',
                'progress:write'
            ],
            'instructor': [
                'courses:read',
                'courses:write',
                'progress:read',
                'students:read'
            ],
            'admin': [
                'courses:*',
                'users:*',
                'analytics:*',
                'settings:*'
            ]
        }
        
        return role_permissions.get(role, [])

# 権限チェックデコレータ
def require_permission(permission: str):
    def decorator(func):
        async def wrapper(
            *args, 
            current_user: UserContext = Depends(AuthManager.get_current_user),
            **kwargs
        ):
            if permission not in current_user.permissions and '*' not in current_user.permissions:
                raise HTTPException(status_code=403, detail="Insufficient permissions")
            return await func(*args, current_user=current_user, **kwargs)
        return wrapper
    return decorator

Phase 2: パターン別実装

2.1 Pattern 1: 学習者ポータル (Flet)

# frontend/src/main.py
import flet as ft
import asyncio
from typing import Optional
from dataclasses import dataclass

from services.auth_service import AuthService
from services.course_service import CourseService
from services.progress_service import ProgressService
from services.realtime_service import RealtimeService
from ui.pages.login_page import LoginPage
from ui.pages.dashboard_page import DashboardPage
from ui.pages.course_page import CoursePage

@dataclass
class AppState:
    current_user: Optional[dict] = None
    current_page: str = "login"
    courses: list = None
    progress: dict = None

class EduConnectApp:
    def __init__(self):
        self.state = AppState()
        self.auth_service = AuthService()
        self.course_service = CourseService()
        self.progress_service = ProgressService()
        self.realtime_service = RealtimeService()
        
    async def main(self, page: ft.Page):
        """メインアプリケーション"""
        
        page.title = "EduConnect - 学習ポータル"
        page.theme_mode = ft.ThemeMode.LIGHT
        page.window_width = 1200
        page.window_height = 800
        
        # ページ状態管理
        self.page = page
        
        # 認証状態確認
        await self._check_auth_status()
        
        # 初期ページ表示
        await self._render_current_page()
        
        # リアルタイム機能初期化
        if self.state.current_user:
            await self._setup_realtime()
    
    async def _check_auth_status(self):
        """認証状態確認"""
        try:
            user = await self.auth_service.get_current_user()
            if user:
                self.state.current_user = user
                self.state.current_page = "dashboard"
                await self._load_user_data()
        except Exception as e:
            print(f"Auth check failed: {e}")
            self.state.current_page = "login"
    
    async def _load_user_data(self):
        """ユーザーデータ読み込み"""
        try:
            # コース一覧取得
            courses = await self.course_service.get_enrolled_courses(
                self.state.current_user['id']
            )
            self.state.courses = courses
            
            # 進捗情報取得
            progress = await self.progress_service.get_user_progress(
                self.state.current_user['id']
            )
            self.state.progress = progress
            
        except Exception as e:
            print(f"Data loading failed: {e}")
    
    async def _render_current_page(self):
        """現在ページのレンダリング"""
        self.page.controls.clear()
        
        if self.state.current_page == "login":
            login_page = LoginPage(self._on_login_success)
            self.page.add(login_page.build())
            
        elif self.state.current_page == "dashboard":
            dashboard_page = DashboardPage(
                user=self.state.current_user,
                courses=self.state.courses,
                progress=self.state.progress,
                on_course_select=self._on_course_select,
                on_logout=self._on_logout
            )
            self.page.add(dashboard_page.build())
            
        elif self.state.current_page == "course":
            course_page = CoursePage(
                course=self.state.selected_course,
                user=self.state.current_user,
                on_back=self._on_back_to_dashboard
            )
            self.page.add(course_page.build())
        
        await self.page.update_async()
    
    async def _setup_realtime(self):
        """リアルタイム機能セットアップ"""
        
        def on_progress_update(payload):
            """進捗更新のリアルタイム同期"""
            asyncio.create_task(self._handle_progress_update(payload))
        
        def on_course_update(payload):
            """コース更新のリアルタイム同期"""
            asyncio.create_task(self._handle_course_update(payload))
        
        # 進捗更新の監視
        await self.realtime_service.subscribe_to_progress_updates(
            self.state.current_user['id'],
            on_progress_update
        )
        
        # コース更新の監視
        await self.realtime_service.subscribe_to_course_updates(
            self.state.current_user['organization_id'],
            on_course_update
        )
    
    async def _handle_progress_update(self, payload):
        """進捗更新ハンドラ"""
        # 進捗データ更新
        await self._load_user_data()
        
        # UI更新
        if self.state.current_page == "dashboard":
            await self._render_current_page()
    
    async def _on_login_success(self, user):
        """ログイン成功ハンドラ"""
        self.state.current_user = user
        self.state.current_page = "dashboard"
        await self._load_user_data()
        await self._setup_realtime()
        await self._render_current_page()
    
    async def _on_course_select(self, course):
        """コース選択ハンドラ"""
        self.state.selected_course = course
        self.state.current_page = "course"
        await self._render_current_page()
    
    async def _on_logout(self):
        """ログアウトハンドラ"""
        await self.auth_service.sign_out()
        await self.realtime_service.cleanup()
        self.state = AppState()
        await self._render_current_page()

# アプリケーション起動
if __name__ == "__main__":
    app = EduConnectApp()
    ft.app(target=app.main, view=ft.AppView.FLET_APP)

2.2 Pattern 2: Edge Functions実装

// edge-functions/course-management/index.ts
import { serve } from "https://deno.land/std@0.208.0/http/server.ts"
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2.39.0'

interface CourseRequest {
  action: 'enroll' | 'progress' | 'complete' | 'certificate'
  courseId: string
  lessonId?: string
  data?: Record<string, any>
}

serve(async (req: Request): Promise<Response> => {
  try {
    // CORS対応
    if (req.method === 'OPTIONS') {
      return new Response(null, {
        status: 200,
        headers: {
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Headers': 'authorization, x-client-info, apikey, content-type',
          'Access-Control-Allow-Methods': 'POST, GET, OPTIONS'
        }
      })
    }

    // 認証確認
    const authHeader = req.headers.get('Authorization')
    if (!authHeader) {
      throw new Error('Missing authorization header')
    }

    const supabase = createClient(
      Deno.env.get('SUPABASE_URL') ?? '',
      Deno.env.get('SUPABASE_SERVICE_ROLE_KEY') ?? ''
    )

    // ユーザー認証
    const { data: { user }, error: authError } = await supabase.auth.getUser(
      authHeader.replace('Bearer ', '')
    )

    if (authError || !user) {
      throw new Error('Invalid authentication')
    }

    // リクエスト処理
    const courseRequest: CourseRequest = await req.json()
    let result

    switch (courseRequest.action) {
      case 'enroll':
        result = await handleCourseEnrollment(supabase, user, courseRequest)
        break
      case 'progress':
        result = await handleProgressUpdate(supabase, user, courseRequest)
        break
      case 'complete':
        result = await handleLessonCompletion(supabase, user, courseRequest)
        break
      case 'certificate':
        result = await handleCertificateGeneration(supabase, user, courseRequest)
        break
      default:
        throw new Error(`Unknown action: ${courseRequest.action}`)
    }

    return new Response(JSON.stringify(result), {
      status: 200,
      headers: {
        'Content-Type': 'application/json',
        'Access-Control-Allow-Origin': '*'
      }
    })

  } catch (error) {
    console.error('Course management error:', error)
    
    return new Response(JSON.stringify({
      success: false,
      error: error.message
    }), {
      status: 500,
      headers: {
        'Content-Type': 'application/json',
        'Access-Control-Allow-Origin': '*'
      }
    })
  }
})

async function handleCourseEnrollment(supabase: any, user: any, request: CourseRequest) {
  // コース情報取得
  const { data: course, error: courseError } = await supabase
    .from('courses')
    .select('*')
    .eq('id', request.courseId)
    .single()

  if (courseError || !course) {
    throw new Error('Course not found')
  }

  // 既存登録確認
  const { data: existingProgress } = await supabase
    .from('learning_progress')
    .select('*')
    .eq('user_id', user.id)
    .eq('course_id', request.courseId)
    .maybeSingle()

  if (existingProgress) {
    return {
      success: true,
      message: 'Already enrolled',
      enrollment: existingProgress
    }
  }

  // 新規登録
  const { data: newProgress, error: enrollError } = await supabase
    .from('learning_progress')
    .insert({
      user_id: user.id,
      course_id: request.courseId,
      completion_percentage: 0,
      time_spent_minutes: 0
    })
    .select()
    .single()

  if (enrollError) {
    throw new Error(`Enrollment failed: ${enrollError.message}`)
  }

  // 分析データ更新
  await updateCourseAnalytics(supabase, request.courseId)

  return {
    success: true,
    message: 'Successfully enrolled',
    enrollment: newProgress
  }
}

async function handleProgressUpdate(supabase: any, user: any, request: CourseRequest) {
  const { courseId, lessonId, data } = request

  // 進捗更新
  const { data: updated, error } = await supabase
    .from('learning_progress')
    .update({
      completion_percentage: data.completionPercentage,
      time_spent_minutes: data.timeSpentMinutes,
      last_accessed_at: new Date().toISOString()
    })
    .eq('user_id', user.id)
    .eq('course_id', courseId)
    .eq('lesson_id', lessonId)
    .select()
    .single()

  if (error) {
    throw new Error(`Progress update failed: ${error.message}`)
  }

  // リアルタイム通知
  await supabase
    .channel('progress_updates')
    .send({
      type: 'broadcast',
      event: 'progress_update',
      payload: {
        userId: user.id,
        courseId,
        lessonId,
        progress: updated
      }
    })

  return {
    success: true,
    progress: updated
  }
}

async function handleCertificateGeneration(supabase: any, user: any, request: CourseRequest) {
  // コース完了確認
  const { data: progress } = await supabase
    .from('learning_progress')
    .select('*')
    .eq('user_id', user.id)
    .eq('course_id', request.courseId)
    .single()

  if (!progress || progress.completion_percentage < 100) {
    throw new Error('Course not completed')
  }

  // 証明書生成API呼び出し
  const certificateResponse = await fetch(`${Deno.env.get('CERTIFICATE_SERVICE_URL')}/generate`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${Deno.env.get('CERTIFICATE_API_KEY')}`
    },
    body: JSON.stringify({
      userId: user.id,
      courseId: request.courseId,
      completionDate: progress.completed_at,
      template: 'standard'
    })
  })

  if (!certificateResponse.ok) {
    throw new Error('Certificate generation failed')
  }

  const certificate = await certificateResponse.json()

  return {
    success: true,
    certificate
  }
}

async function updateCourseAnalytics(supabase: any, courseId: string) {
  // 非同期で分析データ更新
  const { error } = await supabase.rpc('refresh_course_analytics', {
    target_course_id: courseId
  })

  if (error) {
    console.error('Analytics update failed:', error)
  }
}

2.3 Pattern 3: 管理ダッシュボード (FastAPI)

# backend/app/api/v1/analytics.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List, Optional
from datetime import datetime, timedelta

from app.core.auth import AuthManager, UserContext, require_permission
from app.core.database import get_db
from app.services.analytics_service import AnalyticsService
from app.schemas.analytics import (
    CourseAnalytics,
    UserEngagementReport,
    LearningTrendReport
)

router = APIRouter()
auth_manager = AuthManager()

@router.get("/course-performance", response_model=List[CourseAnalytics])
@require_permission("analytics:read")
async def get_course_performance(
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None,
    current_user: UserContext = Depends(auth_manager.get_current_user),
    db: Session = Depends(get_db)
):
    """コースパフォーマンス分析"""
    
    analytics_service = AnalyticsService(db)
    
    # デフォルト期間設定(過去30日)
    if not start_date:
        start_date = datetime.now() - timedelta(days=30)
    if not end_date:
        end_date = datetime.now()
    
    try:
        analytics = await analytics_service.get_course_performance(
            organization_id=current_user.organization_id,
            start_date=start_date,
            end_date=end_date
        )
        
        return analytics
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Analytics generation failed: {str(e)}")

@router.get("/engagement-report", response_model=UserEngagementReport)
@require_permission("analytics:read")
async def get_engagement_report(
    period: str = "30d",
    current_user: UserContext = Depends(auth_manager.get_current_user),
    db: Session = Depends(get_db)
):
    """ユーザーエンゲージメントレポート"""
    
    analytics_service = AnalyticsService(db)
    
    period_mapping = {
        "7d": timedelta(days=7),
        "30d": timedelta(days=30),
        "90d": timedelta(days=90)
    }
    
    if period not in period_mapping:
        raise HTTPException(status_code=400, detail="Invalid period")
    
    try:
        report = await analytics_service.generate_engagement_report(
            organization_id=current_user.organization_id,
            period=period_mapping[period]
        )
        
        return report
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Report generation failed: {str(e)}")

# services/analytics_service.py
import pandas as pd
from sqlalchemy import text
from typing import Dict, List
from datetime import datetime, timedelta

class AnalyticsService:
    def __init__(self, db_session):
        self.db = db_session
    
    async def get_course_performance(
        self, 
        organization_id: str,
        start_date: datetime,
        end_date: datetime
    ) -> List[Dict]:
        """高度なコース分析"""
        
        # 複雑な分析クエリ
        query = text("""
        WITH course_metrics AS (
            SELECT 
                c.id,
                c.title,
                COUNT(DISTINCT lp.user_id) as total_enrollments,
                AVG(lp.completion_percentage) as avg_completion,
                PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY lp.completion_percentage) as median_completion,
                SUM(lp.time_spent_minutes) as total_time_spent,
                COUNT(CASE WHEN lp.completion_percentage = 100 THEN 1 END) as completions,
                
                -- 学習速度分析
                AVG(
                    CASE 
                        WHEN lp.completion_percentage > 0 
                        THEN lp.time_spent_minutes / lp.completion_percentage * 100
                    END
                ) as avg_time_to_complete,
                
                -- エンゲージメント指標
                COUNT(DISTINCT DATE(lp.last_accessed_at)) as active_days,
                AVG(
                    EXTRACT(days FROM lp.updated_at - lp.created_at)
                ) as avg_learning_duration_days
                
            FROM courses c
            LEFT JOIN learning_progress lp ON c.id = lp.course_id
            WHERE c.organization_id = :org_id
                AND lp.created_at BETWEEN :start_date AND :end_date
            GROUP BY c.id, c.title
        ),
        benchmark_data AS (
            SELECT 
                AVG(avg_completion) as industry_avg_completion,
                AVG(total_time_spent / NULLIF(total_enrollments, 0)) as industry_avg_time_per_user
            FROM course_metrics
        )
        SELECT 
            cm.*,
            bd.industry_avg_completion,
            bd.industry_avg_time_per_user,
            
            -- パフォーマンス評価
            CASE 
                WHEN cm.avg_completion > bd.industry_avg_completion * 1.2 THEN 'excellent'
                WHEN cm.avg_completion > bd.industry_avg_completion THEN 'good'
                WHEN cm.avg_completion > bd.industry_avg_completion * 0.8 THEN 'average'
                ELSE 'needs_improvement'
            END as performance_rating,
            
            -- リスク指標
            CASE 
                WHEN cm.avg_completion < 30 AND cm.total_enrollments > 10 THEN 'high_dropout_risk'
                WHEN cm.avg_time_to_complete > bd.industry_avg_time_per_user * 2 THEN 'engagement_risk'
                ELSE 'normal'
            END as risk_indicator
            
        FROM course_metrics cm
        CROSS JOIN benchmark_data bd
        ORDER BY cm.total_enrollments DESC, cm.avg_completion DESC
        """)
        
        result = await self.db.execute(query, {
            'org_id': organization_id,
            'start_date': start_date,
            'end_date': end_date
        })
        
        return [dict(row) for row in result.fetchall()]
    
    async def generate_engagement_report(
        self,
        organization_id: str,
        period: timedelta
    ) -> Dict:
        """ユーザーエンゲージメント詳細分析"""
        
        end_date = datetime.now()
        start_date = end_date - period
        
        # エンゲージメント指標計算
        engagement_query = text("""
        WITH user_activity AS (
            SELECT 
                up.id as user_id,
                up.full_name,
                up.role,
                COUNT(DISTINCT lp.course_id) as courses_enrolled,
                COUNT(DISTINCT DATE(lp.last_accessed_at)) as active_days,
                SUM(lp.time_spent_minutes) as total_time_spent,
                AVG(lp.completion_percentage) as avg_completion,
                MAX(lp.last_accessed_at) as last_activity,
                
                -- セッション分析
                COUNT(DISTINCT 
                    CASE 
                        WHEN lp.last_accessed_at >= :start_date 
                        THEN DATE(lp.last_accessed_at)
                    END
                ) as recent_active_days
                
            FROM user_profiles up
            LEFT JOIN learning_progress lp ON up.id = lp.user_id
            WHERE up.organization_id = :org_id
                AND up.role = 'student'
            GROUP BY up.id, up.full_name, up.role
        ),
        engagement_segments AS (
            SELECT 
                *,
                CASE 
                    WHEN recent_active_days >= 20 AND avg_completion > 70 THEN 'highly_engaged'
                    WHEN recent_active_days >= 10 AND avg_completion > 40 THEN 'moderately_engaged'
                    WHEN recent_active_days >= 3 AND avg_completion > 20 THEN 'low_engaged'
                    ELSE 'at_risk'
                END as engagement_level
            FROM user_activity
        )
        SELECT 
            engagement_level,
            COUNT(*) as user_count,
            AVG(total_time_spent) as avg_time_spent,
            AVG(avg_completion) as avg_completion_rate,
            AVG(courses_enrolled) as avg_courses_enrolled
        FROM engagement_segments
        GROUP BY engagement_level
        ORDER BY 
            CASE engagement_level
                WHEN 'highly_engaged' THEN 1
                WHEN 'moderately_engaged' THEN 2
                WHEN 'low_engaged' THEN 3
                WHEN 'at_risk' THEN 4
            END
        """)
        
        result = await self.db.execute(engagement_query, {
            'org_id': organization_id,
            'start_date': start_date
        })
        
        engagement_data = [dict(row) for row in result.fetchall()]
        
        # 推奨アクション生成
        recommendations = self._generate_engagement_recommendations(engagement_data)
        
        return {
            "period": f"{period.days} days",
            "generated_at": datetime.now().isoformat(),
            "engagement_segments": engagement_data,
            "recommendations": recommendations,
            "summary": {
                "total_users": sum(seg["user_count"] for seg in engagement_data),
                "highly_engaged_percentage": next(
                    (seg["user_count"] for seg in engagement_data if seg["engagement_level"] == "highly_engaged"), 0
                ) / sum(seg["user_count"] for seg in engagement_data) * 100
            }
        }
    
    def _generate_engagement_recommendations(self, engagement_data: List[Dict]) -> List[Dict]:
        """エンゲージメント改善推奨事項生成"""
        
        recommendations = []
        
        for segment in engagement_data:
            level = segment["engagement_level"]
            count = segment["user_count"]
            
            if level == "at_risk" and count > 5:
                recommendations.append({
                    "priority": "high",
                    "target_segment": "at_risk",
                    "action": "implement_re_engagement_campaign",
                    "description": f"{count}名のリスクユーザーに対して再エンゲージメントキャンペーンを実施",
                    "expected_impact": "20-30%のユーザー復帰率"
                })
            
            if level == "low_engaged" and count > 10:
                recommendations.append({
                    "priority": "medium",
                    "target_segment": "low_engaged",
                    "action": "personalized_learning_paths",
                    "description": f"{count}名の低エンゲージメントユーザーに個別化学習パスを提供",
                    "expected_impact": "エンゲージメント40%向上"
                })
        
        return recommendations

Phase 3: 統合・テスト・デプロイメント

3.1 統合テストスイート

# tests/integration/test_full_flow.py
import pytest
import asyncio
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch

from backend.app.main import app
from frontend.src.services.auth_service import AuthService

class TestEduConnectIntegration:
    """エンドツーエンド統合テスト"""
    
    @pytest.fixture
    def client(self):
        return TestClient(app)
    
    @pytest.fixture
    async def authenticated_user(self):
        """テスト用認証ユーザー"""
        auth_service = AuthService()
        return await auth_service.create_test_user({
            "email": "test@example.com",
            "password": "testpassword123",
            "organization_id": "test-org-id",
            "role": "student"
        })
    
    async def test_complete_learning_journey(self, client, authenticated_user):
        """完全な学習ジャーニーテスト"""
        
        # 1. ユーザー登録・ログイン
        login_response = client.post("/api/v1/auth/login", json={
            "email": "test@example.com",
            "password": "testpassword123"
        })
        assert login_response.status_code == 200
        token = login_response.json()["access_token"]
        headers = {"Authorization": f"Bearer {token}"}
        
        # 2. コース一覧取得
        courses_response = client.get("/api/v1/courses", headers=headers)
        assert courses_response.status_code == 200
        courses = courses_response.json()
        assert len(courses) > 0
        
        course_id = courses[0]["id"]
        
        # 3. コース登録 (Edge Function)
        enrollment_response = client.post(
            "/functions/v1/course-management",
            headers=headers,
            json={
                "action": "enroll",
                "courseId": course_id
            }
        )
        assert enrollment_response.status_code == 200
        assert enrollment_response.json()["success"] is True
        
        # 4. 学習進捗更新
        progress_response = client.post(
            "/functions/v1/course-management", 
            headers=headers,
            json={
                "action": "progress",
                "courseId": course_id,
                "lessonId": "lesson-1",
                "data": {
                    "completionPercentage": 50,
                    "timeSpentMinutes": 30
                }
            }
        )
        assert progress_response.status_code == 200
        
        # 5. 進捗確認 (FastAPI)
        progress_check = client.get(f"/api/v1/progress/{course_id}", headers=headers)
        assert progress_check.status_code == 200
        progress_data = progress_check.json()
        assert progress_data["completion_percentage"] == 50
        
        # 6. コース完了
        completion_response = client.post(
            "/functions/v1/course-management",
            headers=headers, 
            json={
                "action": "complete",
                "courseId": course_id,
                "data": {
                    "completionPercentage": 100,
                    "timeSpentMinutes": 120
                }
            }
        )
        assert completion_response.status_code == 200
        
        # 7. 証明書生成
        certificate_response = client.post(
            "/functions/v1/course-management",
            headers=headers,
            json={
                "action": "certificate", 
                "courseId": course_id
            }
        )
        assert certificate_response.status_code == 200
        certificate = certificate_response.json()
        assert "certificate" in certificate
        
        # 8. 分析データ確認 (管理者権限必要)
        admin_token = await self._get_admin_token()
        admin_headers = {"Authorization": f"Bearer {admin_token}"}
        
        analytics_response = client.get(
            "/api/v1/analytics/course-performance",
            headers=admin_headers
        )
        assert analytics_response.status_code == 200
        analytics = analytics_response.json()
        
        # コース完了が分析に反映されているか確認
        course_analytics = next(
            (a for a in analytics if a["id"] == course_id), None
        )
        assert course_analytics is not None
        assert course_analytics["completions"] >= 1
    
    async def test_realtime_synchronization(self, client, authenticated_user):
        """リアルタイム同期テスト"""
        
        # WebSocket接続シミュレーション
        with patch('frontend.src.services.realtime_service.RealtimeService') as mock_realtime:
            mock_client = Mock()
            mock_realtime.return_value = mock_client
            
            # 進捗更新イベント発生
            token = await self._get_user_token(authenticated_user)
            headers = {"Authorization": f"Bearer {token}"}
            
            progress_response = client.post(
                "/functions/v1/course-management",
                headers=headers,
                json={
                    "action": "progress",
                    "courseId": "test-course-id",
                    "data": {"completionPercentage": 75}
                }
            )
            
            # リアルタイムイベントが発火されたか確認
            assert progress_response.status_code == 200
            
            # モックでイベント受信確認
            mock_client.send.assert_called()
            sent_payload = mock_client.send.call_args[0][0]
            assert sent_payload["event"] == "progress_update"
            assert sent_payload["payload"]["progress"]["completion_percentage"] == 75

# パフォーマンステスト
class TestPerformance:
    
    @pytest.mark.performance
    async def test_concurrent_user_load(self):
        """同時接続負荷テスト"""
        
        async def simulate_user_session():
            """ユーザーセッションシミュレーション"""
            # ログイン -> コース閲覧 -> 進捗更新のフロー
            # 実装省略
            pass
        
        # 100同時ユーザーのシミュレーション
        tasks = [simulate_user_session() for _ in range(100)]
        start_time = asyncio.get_event_loop().time()
        
        await asyncio.gather(*tasks)
        
        elapsed_time = asyncio.get_event_loop().time() - start_time
        
        # 性能要件: 100ユーザーセッション完了が30秒以内
        assert elapsed_time < 30.0
        
    @pytest.mark.performance
    async def test_database_query_performance(self, client):
        """データベースクエリ性能テスト"""
        
        admin_token = await self._get_admin_token()
        headers = {"Authorization": f"Bearer {admin_token}"}
        
        start_time = asyncio.get_event_loop().time()
        
        # 重い分析クエリの実行
        response = client.get(
            "/api/v1/analytics/course-performance",
            headers=headers
        )
        
        elapsed_time = asyncio.get_event_loop().time() - start_time
        
        assert response.status_code == 200
        # 性能要件: 分析クエリが2秒以内
        assert elapsed_time < 2.0

3.2 デプロイメント設定

# .github/workflows/deploy.yml
name: EduConnect CI/CD Pipeline

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

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: educonnect

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: test_db
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      
      redis:
        image: redis:7
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Setup Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Setup Deno  
      uses: denoland/setup-deno@v1
      with:
        deno-version: v1.40.x
    
    - name: Install Python dependencies
      run: |
        cd backend
        pip install -r requirements.txt
        pip install pytest pytest-asyncio pytest-cov
    
    - name: Run Python tests
      env:
        DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
        REDIS_URL: redis://localhost:6379
      run: |
        cd backend
        pytest tests/ -v --cov=app --cov-report=xml
    
    - name: Run Deno tests
      run: |
        cd edge-functions
        deno test --allow-all
    
    - name: Run Frontend tests
      run: |
        cd frontend
        python -m pytest tests/ -v
    
    - name: Integration tests
      run: |
        pytest tests/integration/ -v --timeout=300

  security:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    
    - name: Run security scan
      run: |
        pip install safety bandit
        safety check
        bandit -r backend/app/
    
    - name: Dependency vulnerability scan
      uses: pypa/gh-action-pip-audit@v1.0.8

  build-and-deploy:
    needs: [test, security]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Log in to Container Registry
      uses: docker/login-action@v2
      with:
        registry: $
        username: $
        password: $
    
    - name: Build and push Backend
      uses: docker/build-push-action@v4
      with:
        context: ./backend
        push: true
        tags: $/$/backend:latest
    
    - name: Build and push Frontend  
      uses: docker/build-push-action@v4
      with:
        context: ./frontend
        push: true
        tags: $/$/frontend:latest
    
    - name: Deploy Edge Functions
      run: |
        npx supabase functions deploy course-management
        npx supabase functions deploy video-streaming
        npx supabase functions deploy certificate-generator
      env:
        SUPABASE_ACCESS_TOKEN: $
        SUPABASE_PROJECT_ID: $
    
    - name: Deploy to Production
      run: |
        echo "Deploying to production environment"
        # Kubernetes/Docker deployment commands
        kubectl apply -f k8s/production/
      env:
        KUBE_CONFIG: $

3.3 運用監視設定

# infrastructure/monitoring/docker-compose.monitoring.yml
version: '3.8'

services:
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./alerts.yml:/etc/prometheus/alerts.yml
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--web.console.libraries=/etc/prometheus/console_libraries'
      - '--web.console.templates=/etc/prometheus/consoles'
      - '--web.enable-lifecycle'
      - '--web.enable-admin-api'
    
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD}
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
  
  alertmanager:
    image: prom/alertmanager:latest
    ports:
      - "9093:9093"
    volumes:
      - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
  
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    command: -config.file=/etc/loki/local-config.yaml
    volumes:
      - ./loki-config.yaml:/etc/loki/local-config.yaml
  
  promtail:
    image: grafana/promtail:latest
    volumes:
      - /var/log:/var/log:ro
      - ./promtail-config.yaml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml

volumes:
  grafana_data:

最終成果物と検証

成果物一覧

  1. 統合アプリケーション
    • ✅ 3パターン統合EduConnectプラットフォーム
    • ✅ 商用レベルの機能完成度
    • ✅ エンタープライズセキュリティ対応
  2. 運用基盤
    • ✅ 完全なCI/CDパイプライン
    • ✅ 包括的テストスイート
    • ✅ 監視・アラート基盤
  3. ドキュメント
    • ✅ 技術仕様書
    • ✅ 運用手順書
    • ✅ トラブルシューティングガイド

品質検証

# 総合品質チェックスクリプト
#!/bin/bash

echo "🔍 EduConnect品質検証開始"

# 1. コード品質
echo "📊 コード品質チェック"
flake8 backend/app --count --select=E9,F63,F7,F82 --show-source --statistics
black --check backend/app
isort --check-only backend/app

# 2. セキュリティ
echo "🔒 セキュリティスキャン"
bandit -r backend/app
safety check

# 3. テスト
echo "🧪 テスト実行"
pytest backend/tests/ -v --cov=app --cov-report=term-missing
deno test --allow-all edge-functions/

# 4. パフォーマンス
echo "⚡ パフォーマンステスト"
pytest tests/performance/ -v -m performance

# 5. 統合テスト
echo "🔗 統合テスト"
pytest tests/integration/ -v --timeout=300

echo "✅ 品質検証完了"

デプロイメント確認

# 本番デプロイメント確認
curl -f https://api.educonnect.example.com/health || exit 1
curl -f https://app.educonnect.example.com/ || exit 1

echo "🚀 EduConnect統合プラットフォーム デプロイメント完了"

理解度チェック

統合プロジェクト習得確認

  • 3つのアーキテクチャパターンを適切に統合できる
  • ビジネス要件から技術選択を判断できる
  • エンタープライズレベルの品質保証ができる
  • CI/CDパイプラインを構築・運用できる
  • 本番運用監視システムを設計できる

実践課題

  1. 機能拡張: ビデオ会議機能の追加実装
  2. パフォーマンス: 10万ユーザー対応の最適化
  3. 国際化: 多言語・多通貨対応の実装

章間ナビゲーション

前の章 現在の章 次の段階
Chapter 9: スケーリング戦略 Chapter 10: 統合実践プロジェクト 実務適用

学習完了

flowchart LR
    A[Chapter 1-9] --> B[Chapter 10 📍]
    B --> C[実務プロジェクト適用]
    B --> D[チーム開発実践]
    
    style B fill:#e1f5fe

📝 Chapter 10 学習まとめ(最終章)

統合プロジェクトで習得できたスキル

  • ✅ 3つのアーキテクチャパターンの実践的統合設計
  • ✅ 大規模マルチコンポーネントシステムの開発・運用
  • ✅ 企業レベルのプロジェクト管理・チーム協調手法
  • ✅ 商業プロダクトに求められる品質・セキュリティ・パフォーマンス実装

🎯 学習全体の成果

| 学習段階 | 習得内容 | 実践レベル | 適用場面 | |:———|:———|:———–|:———| | Part I | Supabase基礎・認証 | 🌱 個人開発 | 学習・プロトタイプ | | Part II | 3つのアーキテクチャパターン | 🚀 チーム開発 | 実業務・スタートアップ | | Part III | 運用・セキュリティ・パフォーマンス | 💪 企業レベル | エンタープライズ・SaaS | | Part IV | 統合・実践・問題解決 | 🏆 エキスパート | 技術リード・アーキテクト |

🚀 次のステップ(学習修了後の進路)

🔥 即戦力レベル(すぐに活用)

  • ✅ 既存プロジェクトへのSupabase導入提案・実装
  • ✅ 新規プロジェクトでのアーキテクチャ選択・設計責任者
  • ✅ チーム内でのSupabase技術指導・メンタリング

🏗️ 上級エンジニアレベル(更なる発展)

  • ✅ 企業のSupabase移行プロジェクトリード
  • ✅ SaaS製品のテクニカルアーキテクト
  • ✅ Supabaseコミュニティでの技術発信・貢献

🌟 エキスパートレベル(業界貢献)

  • ✅ Supabase関連の技術書籍・記事執筆
  • ✅ 技術カンファレンスでの講演・事例発表
  • ✅ Supabase Partnerships・技術アドバイザー

🎉 学習完了 - おめでとうございます!

あなたは以下の能力を習得しました

  • 🏗️ 設計力: 要件に応じた最適アーキテクチャの選択・設計
  • 🔧 実装力: 3つのパターンすべての実装・統合・運用
  • 🛡️ 品質力: エンタープライズ級のセキュリティ・パフォーマンス実装
  • 🚀 実践力: 実際のビジネス要件を満たすプロダクト開発

Supabaseアーキテクチャパターンの学習が完了しました。実際のプロジェクトでの活用を始めてください!


📍 最終ナビゲーション