はじめに
前回は、社員管理のCRUD（一覧・登録・更新・削除・バリデーション）を実装しました。
前回の記事：Python × Dify × RAGで学ぶ業務システム開発入門【第5回 社員管理CRUD編】
第6回では、研修管理を実装します。
社員管理との大きな違いは次の2点です。
社員管理（第5回）
→ 1テーブル（employee）のCRUD
→ シンプルな登録・更新・削除
研修管理（第6回）
→ 複数テーブルのリレーション（training × employee × training_history）
→ 出席率・理解度・課題スコアの自動集計
→ JOIN を使った複合データの取得
複数テーブルを組み合わせたデータ操作は、実務の業務システムで最も頻繁に登場するパターンです。
本回でしっかり習得してください。
本連載は「作って終わり」ではなく、なぜそう設計するのかを重視しています。
設計の意図を理解することで、他のシステムにも応用できる力が身につきます。
本記事で学ぶこと
| 項目 | 内容 |
|---|---|
| テーブルのリレーション | 外部キーを使った複数テーブルの関連付け |
| JOIN | 複数テーブルを結合してデータを取得する |
| 出席率の自動計算 | 入力データから割合を自動算出する |
| 集計クエリ | AVG・COUNT・ROUND を使ったSQL集計 |
| 受講者登録 | 多対多の関係を中間テーブルで管理する |
リレーションとは
リレーションとは「テーブル間の関係」のことです。
研修管理では以下の関係が存在します。
┌──────────┐ ┌─────────────────────┐ ┌──────────┐
│ employee │ 1 * │ training_history │ * 1 │ training │
│──────────│───────│─────────────────────│───────│──────────│
│ id │ │ employee_id (FK) │ │ id │
│ name │ │ training_id (FK) │ │ name │
│ dept │ │ attendance_rate │ │ start_dt │
└──────────┘ │ understanding_level │ │ end_dt │
│ report_score │ └──────────┘
└─────────────────────┘
関係の読み方
| 関係 | 意味 |
|---|---|
| 1人の社員が複数の研修を受講できる | employee 1 → * training_history |
| 1つの研修に複数の社員が参加できる | training 1 → * training_history |
| training_history が中間テーブル | 多対多の関係を実現する |
なぜ中間テーブルが必要か
NG：employee に training_name カラムを直接持つ
→ 1人が複数の研修を受けると行が増える
→ 社員情報が重複する
→ 正規化に反する
OK：training_history で関係を管理する
→ employee_id と training_id の組み合わせで「誰がどの研修を受けたか」を表現
→ 出席率・理解度も受講ごとに管理できる
実装：training_service.py
研修管理のビジネスロジックを実装します。
# services/training_service.py
from database import get_db
# ============================================================
# 研修マスタ CRUD
# ============================================================
def get_all_trainings() -> list[dict]:
"""研修一覧を取得する（新しい順）"""
conn = get_db()
rows = conn.execute(
"SELECT * FROM training ORDER BY start_date DESC"
).fetchall()
conn.close()
return [dict(row) for row in rows]
def get_training_by_id(training_id: int) -> dict | None:
"""IDで研修を1件取得する"""
conn = get_db()
row = conn.execute(
"SELECT * FROM training WHERE id = ?", (training_id,)
).fetchone()
conn.close()
return dict(row) if row else None
def create_training(data: dict) -> tuple[int, list[str]]:
"""
研修を新規登録する
Returns: (登録した研修のid, エラーメッセージのリスト)
"""
errors = validate_training(data)
if errors:
return -1, errors
conn = get_db()
cur = conn.execute(
"""
INSERT INTO training (training_name, start_date, end_date, description)
VALUES (?, ?, ?, ?)
""",
(
data.get("training_name", "").strip(),
data.get("start_date", "").strip() or None,
data.get("end_date", "").strip() or None,
data.get("description", "").strip() or None,
)
)
conn.commit()
new_id = cur.lastrowid
conn.close()
return new_id, []
def validate_training(data: dict) -> list[str]:
"""研修データのバリデーション"""
errors = []
training_name = data.get("training_name", "").strip()
if not training_name:
errors.append("研修名は必須です")
return errors
# ============================================================
# 受講者管理（training_history）
# ============================================================
def get_training_histories(training_id: int) -> list[dict]:
"""
研修に紐づく受講者一覧と成績を取得する
employee テーブルと JOIN して社員名も含める
"""
conn = get_db()
rows = conn.execute(
"""
SELECT
th.id,
th.employee_id,
th.training_id,
th.attendance_rate,
th.understanding_level,
th.report_score,
e.employee_no,
e.name AS employee_name,
e.department
FROM training_history th
JOIN employee e ON th.employee_id = e.id
WHERE th.training_id = ?
ORDER BY e.employee_no
""",
(training_id,)
).fetchall()
conn.close()
return [dict(row) for row in rows]
def add_trainee(training_id: int, employee_id: int) -> tuple[bool, str]:
"""
研修に受講者を追加する
すでに登録済みの場合はエラーメッセージを返す
Returns:
(成功=True/失敗=False, メッセージ)
"""
conn = get_db()
# 重複チェック
existing = conn.execute(
"""
SELECT id FROM training_history
WHERE training_id = ? AND employee_id = ?
""",
(training_id, employee_id)
).fetchone()
if existing:
conn.close()
return False, "この社員はすでに受講登録されています"
conn.execute(
"""
INSERT INTO training_history (employee_id, training_id)
VALUES (?, ?)
""",
(employee_id, training_id)
)
conn.commit()
conn.close()
return True, "受講者を追加しました"
def update_training_history(history_id: int, data: dict) -> list[str]:
"""
受講履歴（出席率・理解度・課題スコア）を更新する
バリデーション:
- 出席率：0〜100 の数値
- 理解度：0〜100 の整数
- 課題スコア：0〜100 の整数
"""
errors = validate_history(data)
if errors:
return errors
conn = get_db()
conn.execute(
"""
UPDATE training_history
SET attendance_rate = ?,
understanding_level = ?,
report_score = ?
WHERE id = ?
""",
(
float(data.get("attendance_rate", 0)),
int(data.get("understanding_level", 0)),
int(data.get("report_score", 0)),
history_id,
)
)
conn.commit()
conn.close()
return []
def validate_history(data: dict) -> list[str]:
"""受講履歴データのバリデーション"""
errors = []
# 出席率
try:
rate = float(data.get("attendance_rate", 0))
if rate < 0 or rate > 100:
errors.append("出席率は 0〜100 の範囲で入力してください")
except (ValueError, TypeError):
errors.append("出席率は数値で入力してください")
# 理解度
try:
level = int(data.get("understanding_level", 0))
if level < 0 or level > 100:
errors.append("理解度は 0〜100 の範囲で入力してください")
except (ValueError, TypeError):
errors.append("理解度は整数で入力してください")
# 課題スコア
try:
score = int(data.get("report_score", 0))
if score < 0 or score > 100:
errors.append("課題スコアは 0〜100 の範囲で入力してください")
except (ValueError, TypeError):
errors.append("課題スコアは整数で入力してください")
return errors
出席率の自動計算
出席率は画面から直接入力する方式と、日別出席データから自動計算する方式があります。
本システムでは学習目的で直接入力方式を採用しますが、自動計算のロジックも紹介します。
# 自動計算方式の例（参考）
def calc_attendance_rate(attended_days: int, total_days: int) -> float:
"""
出席率を計算する
Args:
attended_days: 出席日数
total_days: 研修全日数
Returns:
出席率（0.0〜100.0）
total_days が 0 の場合は 0.0 を返す
"""
if total_days == 0:
return 0.0
return round(attended_days / total_days * 100, 1)
# 使用例
rate = calc_attendance_rate(attended_days=9, total_days=10)
# → 90.0
集計クエリ
研修ごとの平均出席率・平均理解度・平均課題スコアをSQLで集計します。
def get_training_summary(training_id: int) -> dict:
"""
研修の集計結果を取得する
- 受講者数
- 平均出席率
- 平均理解度
- 平均課題スコア
"""
conn = get_db()
row = conn.execute(
"""
SELECT
COUNT(*) AS trainee_count,
ROUND(AVG(attendance_rate), 1) AS avg_attendance,
ROUND(AVG(understanding_level), 1) AS avg_understanding,
ROUND(AVG(report_score), 1) AS avg_report_score
FROM training_history
WHERE training_id = ?
""",
(training_id,)
).fetchone()
conn.close()
return dict(row) if row else {}
SQLの集計関数
| 関数 | 意味 | 例 |
|---|---|---|
COUNT(*) |
行数を数える | 受講者数 |
AVG(col) |
平均値を計算する | 平均出席率 |
ROUND(val, n) |
小数点n桁に丸める | 90.333... → 90.3 |
SUM(col) |
合計値を計算する | （今回は未使用） |
JOINの仕組み
training_history テーブルには employee_id（数値）しか入っていません。
社員名や部署名を一緒に取得するために JOIN を使います。
-- JOIN なし（employee_id しか取れない）
SELECT * FROM training_history WHERE training_id = 1;
-- → employee_id=1, attendance_rate=90.0, ...
-- JOIN あり（社員名・部署も取得できる）
SELECT
th.attendance_rate,
th.understanding_level,
e.name AS employee_name,
e.department
FROM training_history th
JOIN employee e ON th.employee_id = e.id
WHERE th.training_id = 1;
-- → employee_name="山田太郎", department="人事部", attendance_rate=90.0, ...
JOINの動きを図で理解する
training_history employee
┌────────────┬────┐ ┌────┬──────────┐
│employee_id │ ...│ │ id │ name │
├────────────┼────┤ ├────┼──────────┤
│ 1 │ ...│──────▶│ 1 │ 山田太郎 │
│ 2 │ ...│──────▶│ 2 │ 鈴木花子 │
│ 3 │ ...│──────▶│ 3 │ 田中一郎 │
└────────────┴────┘ └────┴──────────┘
ON th.employee_id = e.id
JOIN は「共通のキーを使って2つのテーブルを横に結合する」操作です。
実装：app.py（研修管理ルーティング）
# app.py（研修管理部分）
from services.training_service import (
get_all_trainings, get_training_by_id, create_training,
get_training_histories, add_trainee,
update_training_history, get_training_summary
)
from services.employee_service import get_all_employees
@app.route("/trainings")
@login_required
@roles_required("admin", "staff")
def training_list():
"""研修一覧"""
trainings = get_all_trainings()
return render_template("training_list.html", trainings=trainings)
@app.route("/trainings/new", methods=["GET", "POST"])
@login_required
@roles_required("admin", "staff")
def training_new():
"""研修登録"""
if request.method == "POST":
new_id, errors = create_training(request.form)
if errors:
for err in errors:
flash(err, "error")
return render_template("training_form.html",
training=request.form)
flash("研修を登録しました", "success")
return redirect(url_for("training_list"))
return render_template("training_form.html", training=None)
@app.route("/trainings/<int:training_id>/history", methods=["GET", "POST"])
@login_required
@roles_required("admin", "staff")
def training_history(training_id):
"""研修履歴（受講者一覧・成績入力）"""
training = get_training_by_id(training_id)
if training is None:
flash("研修が見つかりません", "error")
return redirect(url_for("training_list"))
# 受講者追加
if request.method == "POST" and "add_employee_id" in request.form:
employee_id = int(request.form["add_employee_id"])
success, message = add_trainee(training_id, employee_id)
flash(message, "success" if success else "error")
return redirect(url_for("training_history",
training_id=training_id))
# 成績更新
if request.method == "POST" and "history_id" in request.form:
history_id = int(request.form["history_id"])
errors = update_training_history(history_id, request.form)
if errors:
for err in errors:
flash(err, "error")
else:
flash("成績を更新しました", "success")
return redirect(url_for("training_history",
training_id=training_id))
# 一覧データ取得
histories = get_training_histories(training_id)
summary = get_training_summary(training_id)
employees = get_all_employees() # 受講者追加用の社員リスト
return render_template(
"training_history.html",
training=training,
histories=histories,
summary=summary,
employees=employees
)
実装：テンプレート（training_history.html）
研修の受講者一覧・成績入力・集計結果を1画面で表示します。
{% extends "base.html" %}
{% block title %}研修履歴 - {{ training.training_name }}{% endblock %}
{% block content %}
<div class="page-header">
<h1>📚 {{ training.training_name }}</h1>
<a href="{{ url_for('training_list') }}" class="btn btn-secondary">← 研修一覧</a>
</div>
<!-- 研修情報 -->
<div class="info-card">
<p>期間：{{ training.start_date or "未定" }} 〜 {{ training.end_date or "未定" }}</p>
<p>{{ training.description or "" }}</p>
</div>
<!-- 集計結果 -->
{% if summary and summary.trainee_count > 0 %}
<div class="summary-card">
<h3>📊 集計結果</h3>
<table class="table table-compact">
<tr>
<th>受講者数</th>
<th>平均出席率</th>
<th>平均理解度</th>
<th>平均課題スコア</th>
</tr>
<tr>
<td>{{ summary.trainee_count }} 名</td>
<td>{{ summary.avg_attendance }}%</td>
<td>{{ summary.avg_understanding }}点</td>
<td>{{ summary.avg_report_score }}点</td>
</tr>
</table>
</div>
{% endif %}
<!-- 受講者追加フォーム -->
<div class="form-card">
<h3>受講者を追加</h3>
<form method="POST" class="inline-form">
<select name="add_employee_id" required>
<option value="">-- 社員を選択 --</option>
{% for emp in employees %}
<option value="{{ emp.id }}">
{{ emp.employee_no }} - {{ emp.name }}（{{ emp.department or "―" }}）
</option>
{% endfor %}
</select>
<button type="submit" class="btn btn-primary">追加</button>
</form>
</div>
<!-- 受講者一覧・成績入力 -->
<h3>受講者一覧</h3>
<table class="table">
<thead>
<tr>
<th>社員番号</th>
<th>氏名</th>
<th>部署</th>
<th>出席率（%）</th>
<th>理解度</th>
<th>課題スコア</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% if histories %}
{% for h in histories %}
<tr>
<form method="POST">
<input type="hidden" name="history_id" value="{{ h.id }}">
<td>{{ h.employee_no }}</td>
<td>{{ h.employee_name }}</td>
<td>{{ h.department or "―" }}</td>
<td>
<input type="number" name="attendance_rate"
value="{{ h.attendance_rate }}"
min="0" max="100" step="0.1" style="width:80px">
</td>
<td>
<input type="number" name="understanding_level"
value="{{ h.understanding_level }}"
min="0" max="100" style="width:80px">
</td>
<td>
<input type="number" name="report_score"
value="{{ h.report_score }}"
min="0" max="100" style="width:80px">
</td>
<td>
<button type="submit" class="btn btn-small btn-primary">保存</button>
</td>
</form>
</tr>
{% endfor %}
{% else %}
<tr>
<td colspan="7" class="text-center text-muted">
受講者が登録されていません。上のフォームから追加してください
</td>
</tr>
{% endif %}
</tbody>
</table>
{% endblock %}
実装：テンプレート（training_list.html）
{% extends "base.html" %}
{% block title %}研修一覧 - AI人事・研修システム{% endblock %}
{% block content %}
<div class="page-header">
<h1>📚 研修管理</h1>
<a href="{{ url_for('training_new') }}" class="btn btn-primary">＋ 研修登録</a>
</div>
<table class="table">
<thead>
<tr>
<th>研修名</th>
<th>開始日</th>
<th>終了日</th>
<th>概要</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% if trainings %}
{% for tr in trainings %}
<tr>
<td>{{ tr.training_name }}</td>
<td>{{ tr.start_date or "未定" }}</td>
<td>{{ tr.end_date or "未定" }}</td>
<td>{{ tr.description or "―" }}</td>
<td>
<a href="{{ url_for('training_history', training_id=tr.id) }}"
class="btn btn-small btn-secondary">受講者・成績</a>
</td>
</tr>
{% endfor %}
{% else %}
<tr>
<td colspan="5" class="text-center text-muted">
研修データがありません。「研修登録」から追加してください
</td>
</tr>
{% endif %}
</tbody>
</table>
{% endblock %}
実装：テンプレート（training_form.html）
{% extends "base.html" %}
{% block title %}研修登録 - AI人事・研修システム{% endblock %}
{% block content %}
<div class="page-header">
<h1>📚 研修登録</h1>
</div>
<form method="POST" class="form-card">
<div class="form-group">
<label for="training_name">研修名 <span class="required">*</span></label>
<input type="text" id="training_name" name="training_name"
value="{{ training.training_name if training else '' }}"
placeholder="例: Python基礎研修" required>
</div>
<div class="form-group">
<label for="start_date">開始日</label>
<input type="date" id="start_date" name="start_date"
value="{{ training.start_date if training else '' }}">
</div>
<div class="form-group">
<label for="end_date">終了日</label>
<input type="date" id="end_date" name="end_date"
value="{{ training.end_date if training else '' }}">
</div>
<div class="form-group">
<label for="description">研修概要</label>
<textarea id="description" name="description" rows="3"
placeholder="研修の目的や内容を入力">{{ training.description if training else '' }}</textarea>
</div>
<div class="form-actions">
<button type="submit" class="btn btn-primary">登録する</button>
<a href="{{ url_for('training_list') }}" class="btn btn-secondary">キャンセル</a>
</div>
</form>
{% endblock %}
社員管理CRUDとの比較
第5回と第6回の実装パターンを比較して、共通点と違いを整理します。
| 観点 | 社員管理（第5回） | 研修管理（第6回） |
|---|---|---|
| テーブル数 | 1テーブル（employee） | 3テーブル（training, employee, training_history） |
| 主なSQL | SELECT / INSERT / UPDATE | SELECT + JOIN / INSERT / UPDATE |
| データ取得 | 単一テーブルから取得 | JOIN で複数テーブルを結合して取得 |
| バリデーション | 必須チェック・重複チェック・形式チェック | 数値範囲チェック（0〜100）を追加 |
| 表示 | 静的な一覧表示 | 集計結果（AVG / COUNT）を表示 |
| フォーム | 登録・編集が別アクション | 一覧画面内にインライン編集フォーム |
共通している設計パターン
1. 3層構造を維持する
app.py → service → database
2. バリデーションはサービス層で行う
フォームデータをそのまま INSERT しない
3. プレースホルダ（?）を徹底する
SQLインジェクション対策
4. flash メッセージで操作結果を通知する
成功 / エラーをユーザーにフィードバック
動作確認
手順
# Flaskアプリを起動
python app.py
# ブラウザで http://localhost:5000/trainings にアクセス
確認ポイント
| 操作 | 期待する動作 |
|---|---|
| 研修一覧を開く | 登録済みの研修が一覧表示される |
| 研修を新規登録する | 研修名・期間を入力して登録。一覧に追加される |
| 研修名を空にして送信 | 「研修名は必須です」エラーが表示される |
| 「受講者・成績」を開く | 研修の詳細画面が表示される |
| 受講者をプルダウンから選択して追加 | 受講者一覧に追加される |
| 同じ社員を2回追加する | 「すでに受講登録されています」エラーが表示される |
| 出席率に「105」を入力して保存 | 「0〜100 の範囲で」エラーが表示される |
| 正しいスコアを入力して保存 | 「成績を更新しました」と表示。集計結果が更新される |
| 集計結果を確認する | 受講者数・平均出席率・平均理解度・平均課題スコアが表示される |
今後の連載予定
| 回 | タイトル | 主な内容 |
|---|---|---|
| 第1回 | 業務システム全体設計編 | システム概要・技術選定・アーキテクチャ |
| 第2回 | 要求定義・要件定義編 | 業務分析・機能要件・非機能要件 |
| 第3回 | ER図・画面設計編 | テーブル設計・画面遷移図・ワイヤーフレーム |
| 第4回 | Flaskログイン機能編 | セッション・パスワードハッシュ・認証ミドルウェア |
| 第5回 | 社員管理CRUD編 | 一覧・登録・更新・削除・バリデーション |
| 第6回 | 研修管理編（本記事） | リレーション・集計・出席率自動計算 |
| 第7回 | Excel業務自動化編 | pandas取込・openpyxl出力・テンプレート活用 |
| 第8回 | Dify API連携編 | プロンプト設計・API呼び出し・エラーハンドリング |
| 第9回 | RAG構築編 | Knowledgeへの登録・Embedding・検索精度改善 |
| 第10回 | FAQチャットボット編 | チャットUI・会話履歴・ストリーミングレスポンス |
| 第11回 | GitHubチーム開発編 | ブランチ戦略・Pull Request・コードレビュー |
| 第12回 | テスト・振り返り編 | pytest・テスト設計・デモ・振り返り |
おわりに
第6回では、複数テーブルのリレーションと集計を使った研修管理を実装しました。
ポイントを振り返ります。
-
リレーション：
training_historyを中間テーブルとして、社員と研修の多対多の関係を実現した -
JOIN：
employeeテーブルと結合して、社員名・部署を含む受講者一覧を取得した -
集計クエリ：
AVG/COUNT/ROUNDで平均出席率・平均理解度を計算した - バリデーション：数値範囲チェック（0〜100）を追加し、不正なスコア入力を防いだ
- パターンの共通化：3層構造・flash通知・プレースホルダは社員管理と同じ設計パターンを踏襲した
次回は「Excel業務自動化編」として、pandas でのExcelインポートと openpyxl での帳票出力を実装します。
研修管理で登録したデータを Excel から一括取り込みする機能を追加します。
次回 : Python × Dify × RAGで学ぶ業務システム開発入門【第7回 Excel業務自動化編】