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:
最終成果物と検証
成果物一覧
- 統合アプリケーション
- ✅ 3パターン統合EduConnectプラットフォーム
- ✅ 商用レベルの機能完成度
- ✅ エンタープライズセキュリティ対応
- 運用基盤
- ✅ 完全なCI/CDパイプライン
- ✅ 包括的テストスイート
- ✅ 監視・アラート基盤
- ドキュメント
- ✅ 技術仕様書
- ✅ 運用手順書
- ✅ トラブルシューティングガイド
品質検証
# 総合品質チェックスクリプト
#!/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パイプラインを構築・運用できる
- 本番運用監視システムを設計できる
実践課題
- 機能拡張: ビデオ会議機能の追加実装
- パフォーマンス: 10万ユーザー対応の最適化
- 国際化: 多言語・多通貨対応の実装
章間ナビゲーション
前の章 | 現在の章 | 次の段階 |
---|---|---|
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アーキテクチャパターンの学習が完了しました。実際のプロジェクトでの活用を始めてください!
📍 最終ナビゲーション