はじめに
前回は、Flaskのログイン機能（パスワードハッシュ化・セッション管理・認証デコレータ）を実装しました。
前回の記事：Python × Dify × RAGで学ぶ業務システム開発入門【第4回 Flaskログイン機能編】
第5回では、業務システムの中核となる社員管理のCRUDを実装します。
CRUDとは、データ操作の基本4操作のことです。
C：Create → 社員登録
R：Read → 社員一覧・検索・詳細表示
U：Update → 社員情報更新
D：Delete → 社員削除（論理削除）
CRUDが実装できれば、研修管理・評価管理など他のあらゆる機能も同じパターンで作れます。
本回でそのパターンをしっかり習得してください。
本連載は「作って終わり」ではなく、なぜそう設計するのかを重視しています。
設計の意図を理解することで、他のシステムにも応用できる力が身につきます。
本記事で学ぶこと
| 項目 | 内容 |
|---|---|
| サービス層の実装 | DBアクセスをサービスクラスに分離する |
| 一覧・検索 | SQLの WHERE・LIKE を使った絞り込み |
| 登録・更新 | フォームデータの受け取りとDB保存 |
| 論理削除 | レコードを消さずに無効化する設計 |
| バリデーション | 不正な入力データをサーバー側で検証する |
CRUDの実装パターン
本システムでは、すべてのCRUD操作を以下の3層に分けて実装します。
app.py（ルーティング層）
↕ フォームデータの受け渡し
services/employee_service.py（サービス層）
↕ SQL実行・データ変換
database.db（データ層）
| 層 | ファイル | 役割 |
|---|---|---|
| ルーティング層 | app.py |
URLとHTTPメソッドの定義、テンプレートへのデータ渡し |
| サービス層 | services/employee_service.py |
ビジネスロジック、DB操作、バリデーション |
| データ層 | database.db |
データの永続化 |
この分離により、「画面の変更」と「業務ロジックの変更」を独立して行えます。
実装：employee_service.py
サービス層からすべての社員CRUD操作を実装します。
# services/employee_service.py
from database import get_db
# ============================================================
# Read：一覧・検索・詳細取得
# ============================================================
def get_all_employees(keyword: str = "") -> list[dict]:
"""
社員一覧を取得する
keyword が指定された場合は氏名・部署・役職で部分一致検索
退職者（is_active=0）は除外する
"""
conn = get_db()
if keyword:
rows = conn.execute(
"""
SELECT * FROM employee
WHERE is_active = 1
AND (name LIKE ?
OR department LIKE ?
OR position LIKE ?)
ORDER BY employee_no
""",
(f"%{keyword}%", f"%{keyword}%", f"%{keyword}%")
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM employee WHERE is_active = 1 ORDER BY employee_no"
).fetchall()
conn.close()
return [dict(row) for row in rows]
def get_employee_by_id(employee_id: int) -> dict | None:
"""
IDで社員を1件取得する（退職者も含む）
見つからない場合は None を返す
"""
conn = get_db()
row = conn.execute(
"SELECT * FROM employee WHERE id = ?", (employee_id,)
).fetchone()
conn.close()
return dict(row) if row else None
def get_employee_by_no(employee_no: str) -> dict | None:
"""社員番号で社員を1件取得する"""
conn = get_db()
row = conn.execute(
"SELECT * FROM employee WHERE employee_no = ?", (employee_no,)
).fetchone()
conn.close()
return dict(row) if row else None
# ============================================================
# バリデーション
# ============================================================
def validate_employee(data: dict, is_update: bool = False) -> list[str]:
"""
社員データのバリデーション
エラーメッセージのリストを返す（空リストならバリデーション通過）
Args:
data: フォームから受け取った dict
is_update: True の場合は社員番号の重複チェックをスキップ
"""
errors = []
# 社員番号：必須・重複チェック
employee_no = data.get("employee_no", "").strip()
if not employee_no:
errors.append("社員番号は必須です")
elif not is_update:
existing = get_employee_by_no(employee_no)
if existing:
errors.append(f"社員番号 {employee_no} はすでに登録されています")
# 氏名：必須
name = data.get("name", "").strip()
if not name:
errors.append("氏名は必須です")
# 入社日：日付フォーマットチェック（入力がある場合のみ）
hire_date = data.get("hire_date", "").strip()
if hire_date:
import re
if not re.match(r"^\d{4}-\d{2}-\d{2}$", hire_date):
errors.append("入社日は YYYY-MM-DD 形式で入力してください")
return errors
# ============================================================
# Create：登録
# ============================================================
def create_employee(data: dict) -> tuple[int, list[str]]:
"""
社員を新規登録する
Returns:
(登録した社員のid, エラーメッセージのリスト)
エラーがある場合は id=-1 を返す
"""
errors = validate_employee(data, is_update=False)
if errors:
return -1, errors
conn = get_db()
cur = conn.execute(
"""
INSERT INTO employee
(employee_no, name, department, position, hire_date)
VALUES (?, ?, ?, ?, ?)
""",
(
data.get("employee_no", "").strip(),
data.get("name", "").strip(),
data.get("department", "").strip() or None,
data.get("position", "").strip() or None,
data.get("hire_date", "").strip() or None,
)
)
conn.commit()
new_id = cur.lastrowid
conn.close()
return new_id, []
# ============================================================
# Update：更新
# ============================================================
def update_employee(employee_id: int, data: dict) -> list[str]:
"""
社員情報を更新する
Returns:
エラーメッセージのリスト（空リストなら更新成功）
"""
errors = validate_employee(data, is_update=True)
if errors:
return errors
conn = get_db()
conn.execute(
"""
UPDATE employee
SET name = ?,
department = ?,
position = ?,
hire_date = ?,
updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(
data.get("name", "").strip(),
data.get("department", "").strip() or None,
data.get("position", "").strip() or None,
data.get("hire_date", "").strip() or None,
employee_id,
)
)
conn.commit()
conn.close()
return []
# ============================================================
# Delete：論理削除
# ============================================================
def delete_employee(employee_id: int) -> bool:
"""
社員を論理削除する（is_active = 0 に更新）
物理削除ではなく is_active フラグを 0 にするだけ
研修履歴・評価データに影響を与えない
Returns:
True: 削除成功 / False: 対象が見つからない
"""
employee = get_employee_by_id(employee_id)
if employee is None:
return False
conn = get_db()
conn.execute(
"UPDATE employee SET is_active = 0, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(employee_id,)
)
conn.commit()
conn.close()
return True
論理削除とは
削除には「物理削除」と「論理削除」の2種類があります。
物理削除（DELETE文）
→ レコードをDBから完全に消す
→ 問題：研修履歴・評価データの employee_id が
参照先を失い、データの整合性が壊れる
論理削除（is_active フラグ）
→ is_active = 0 に更新するだけでレコードは残す
→ 利点①：退職後も研修履歴・評価は参照できる
→ 利点②：誤って削除した場合に復元できる
→ 利点③：「いつ退職したか」の履歴が残る
一覧表示では WHERE is_active = 1 を付けるだけで退職者を除外できます。
実装：app.py（社員管理ルーティング）
# app.py（社員管理部分）
from flask import render_template, request, redirect, url_for, flash
from auth import login_required, roles_required
from services.employee_service import (
get_all_employees, get_employee_by_id,
create_employee, update_employee, delete_employee
)
@app.route("/employees")
@login_required
@roles_required("admin", "staff")
def employee_list():
"""社員一覧・検索"""
keyword = request.args.get("keyword", "").strip()
employees = get_all_employees(keyword=keyword)
return render_template(
"employee_list.html",
employees=employees,
keyword=keyword
)
@app.route("/employees/new", methods=["GET", "POST"])
@login_required
@roles_required("admin", "staff")
def employee_new():
"""社員登録"""
if request.method == "POST":
new_id, errors = create_employee(request.form)
if errors:
# バリデーションエラー：入力値を保持してフォームを再表示
for err in errors:
flash(err, "error")
return render_template("employee_form.html",
employee=request.form,
mode="new")
flash("社員を登録しました", "success")
return redirect(url_for("employee_list"))
return render_template("employee_form.html", employee=None, mode="new")
@app.route("/employees/<int:employee_id>/edit", methods=["GET", "POST"])
@login_required
@roles_required("admin", "staff")
def employee_edit(employee_id):
"""社員編集"""
employee = get_employee_by_id(employee_id)
if employee is None:
flash("社員が見つかりません", "error")
return redirect(url_for("employee_list"))
if request.method == "POST":
errors = update_employee(employee_id, request.form)
if errors:
for err in errors:
flash(err, "error")
return render_template("employee_form.html",
employee=request.form,
mode="edit")
flash("社員情報を更新しました", "success")
return redirect(url_for("employee_list"))
return render_template("employee_form.html",
employee=employee,
mode="edit")
@app.route("/employees/<int:employee_id>/delete", methods=["POST"])
@login_required
@roles_required("admin")
def employee_delete(employee_id):
"""社員削除（論理削除）"""
success = delete_employee(employee_id)
if success:
flash("社員を削除しました", "success")
else:
flash("社員が見つかりません", "error")
return redirect(url_for("employee_list"))
削除は GET ではなく POST メソッドで受け付けます。
GET で削除できる設計にすると、URLを誰かに送るだけで意図せず削除されるリスクがあります。
HTMLフォームの method="POST" と onsubmit="return confirm(...)" で安全に実装します。
実装：テンプレート（employee_list.html）
{% extends "base.html" %}
{% block title %}社員一覧 - AI人事・研修システム{% endblock %}
{% block content %}
<div class="page-header">
<h1>👤 社員管理</h1>
<a href="{{ url_for('employee_new') }}" class="btn btn-primary">＋ 新規登録</a>
</div>
<!-- 検索フォーム -->
<form method="GET" action="{{ url_for('employee_list') }}" class="search-form">
<input type="text" name="keyword"
placeholder="氏名・部署・役職で検索"
value="{{ keyword }}">
<button type="submit" class="btn btn-secondary">検索</button>
{% if keyword %}
<a href="{{ url_for('employee_list') }}" class="btn btn-link">クリア</a>
{% endif %}
</form>
{% if keyword %}
<p class="search-result">「{{ keyword }}」の検索結果：{{ employees|length }} 件</p>
{% endif %}
<!-- 社員一覧テーブル -->
<table class="table">
<thead>
<tr>
<th>社員番号</th>
<th>氏名</th>
<th>部署</th>
<th>役職</th>
<th>入社日</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% if employees %}
{% for emp in employees %}
<tr>
<td>{{ emp.employee_no }}</td>
<td>{{ emp.name }}</td>
<td>{{ emp.department or "―" }}</td>
<td>{{ emp.position or "―" }}</td>
<td>{{ emp.hire_date or "―" }}</td>
<td>
<a href="{{ url_for('employee_edit', employee_id=emp.id) }}"
class="btn btn-small btn-secondary">編集</a>
<!-- 削除はPOSTで送信し、確認ダイアログを表示する -->
<form method="POST"
action="{{ url_for('employee_delete', employee_id=emp.id) }}"
style="display:inline;"
onsubmit="return confirm('{{ emp.name }} を削除しますか？\nこの操作は取り消せません。');">
<button type="submit" class="btn btn-small btn-danger">削除</button>
</form>
</td>
</tr>
{% endfor %}
{% else %}
<tr>
<td colspan="6" class="text-center text-muted">
{% if keyword %}
「{{ keyword }}」に一致する社員は見つかりませんでした
{% else %}
社員データがありません。「新規登録」から追加してください
{% endif %}
</td>
</tr>
{% endif %}
</tbody>
</table>
{% endblock %}
実装：テンプレート（employee_form.html）
登録と編集で同じフォームを使い回します。
mode 変数で「登録」か「編集」かを判定します。
{% extends "base.html" %}
{% block title %}
{% if mode == "edit" %}社員編集{% else %}社員登録{% endif %} - AI人事・研修システム
{% endblock %}
{% block content %}
<div class="page-header">
<h1>👤 {% if mode == "edit" %}社員編集{% else %}社員登録{% endif %}</h1>
</div>
<form method="POST" class="form-card">
<div class="form-group">
<label for="employee_no">社員番号 <span class="required">*</span></label>
<input type="text" id="employee_no" name="employee_no"
value="{{ employee.employee_no if employee else '' }}"
placeholder="例: EMP001"
{% if mode == "edit" %}readonly{% endif %}
required>
{% if mode == "edit" %}
<small class="text-muted">社員番号は変更できません</small>
{% endif %}
</div>
<div class="form-group">
<label for="name">氏名 <span class="required">*</span></label>
<input type="text" id="name" name="name"
value="{{ employee.name if employee else '' }}"
placeholder="例: 山田 太郎" required>
</div>
<div class="form-group">
<label for="department">部署</label>
<input type="text" id="department" name="department"
value="{{ employee.department if employee else '' }}"
placeholder="例: 人事部">
</div>
<div class="form-group">
<label for="position">役職</label>
<input type="text" id="position" name="position"
value="{{ employee.position if employee else '' }}"
placeholder="例: 主任">
</div>
<div class="form-group">
<label for="hire_date">入社日</label>
<input type="date" id="hire_date" name="hire_date"
value="{{ employee.hire_date if employee else '' }}">
</div>
<div class="form-actions">
<button type="submit" class="btn btn-primary">
{% if mode == "edit" %}更新する{% else %}登録する{% endif %}
</button>
<a href="{{ url_for('employee_list') }}" class="btn btn-secondary">
キャンセル
</a>
</div>
</form>
{% endblock %}
バリデーションの考え方
バリデーションとは「入力データが正しい形式かどうかを検証する処理」です。
なぜサーバー側でバリデーションが必要か
HTMLの required 属性はブラウザ側の検証
→ DevToolsで属性を削除すれば簡単に回避できる
→ APIを直接叩けば完全にスキップできる
サーバー側のバリデーションは回避できない
→ どんな方法で送られてきても検証できる
→ セキュリティの最後の砦
本システムのバリデーションルール
| フィールド | ルール |
|---|---|
| 社員番号 | 必須・空白不可・新規登録時のみ重複チェック |
| 氏名 | 必須・空白不可 |
| 部署 | 任意（空でも可） |
| 役職 | 任意（空でも可） |
| 入社日 | 任意・入力する場合は YYYY-MM-DD 形式 |
バリデーションエラー時の処理フロー
POST /employees/new
↓
validate_employee(data) 実行
↓ エラーあり
flash でエラーメッセージを追加
入力値を保持したままフォームを再表示（redirect しない）
↓ エラーなし
DBに登録
flash で成功メッセージを追加
一覧画面へ redirect
バリデーションエラー時は redirect ではなく render_template で同じ画面を返します。
redirect するとフォームの入力値が消えてしまい、ユーザーが再入力しなければならなくなるためです。
動作確認
手順
# Flaskアプリを起動
python app.py
# ブラウザで http://localhost:5000/employees にアクセス
確認ポイント
| 操作 | 期待する動作 |
|---|---|
| 社員一覧を開く | 登録済み社員が一覧表示される |
| 検索ボックスに「人事」と入力 | 人事部の社員のみ表示される |
| 新規登録：社員番号を空にして送信 | 「社員番号は必須です」エラーが表示される |
| 新規登録：重複する社員番号で送信 | 「すでに登録されています」エラーが表示される |
| 正しい値で新規登録 | 一覧に追加され、成功メッセージが表示される |
| 社員編集：氏名を空にして送信 | 「氏名は必須です」エラーが表示される |
| 社員削除：確認ダイアログで「OK」 | 一覧から消え、成功メッセージが表示される |
| 削除した社員が一覧に表示されない |
is_active=0 として論理削除されている |
SQLインジェクション対策
本実装では ? プレースホルダを使っているため、SQLインジェクションを防いでいます。
# NG：文字列結合（SQLインジェクションに脆弱）
keyword = request.args.get("keyword")
rows = conn.execute(
f"SELECT * FROM employee WHERE name LIKE '%{keyword}%'"
).fetchall()
# keyword に "' OR '1'='1" を入れると全件取得される
# OK：プレースホルダを使用
rows = conn.execute(
"SELECT * FROM employee WHERE name LIKE ?",
(f"%{keyword}%",) # ← タプルで渡す
).fetchall()
# keyword がどんな値でも、文字列として安全に扱われる
SQLを組み立てるときは絶対に文字列結合を使わないでください。
プレースホルダ（?）を使えば、ユーザーの入力はSQL命令として解釈されません。
今後の連載予定
| 回 | タイトル | 主な内容 |
|---|---|---|
| 第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・テスト設計・デモ・振り返り |
おわりに
第5回では、社員管理のCRUD全操作を実装しました。
ポイントを振り返ります。
-
サービス層の分離：
employee_service.pyにDBアクセスとビジネスロジックを集約し、app.pyはルーティングだけを担当した -
一覧・検索：
LIKE句と%keyword%で部分一致検索を実装した -
論理削除：
is_active = 0フラグで退職者データを残しつつ一覧から除外した - バリデーション：サーバー側でエラーチェックし、エラー時は入力値を保持してフォームを再表示した
-
SQLインジェクション対策：プレースホルダ（
?）を徹底してSQLインジェクションを防いだ
このCRUDパターンは研修管理・評価管理でもまったく同じ形で使えます。
次回の「研修管理編」では、複数テーブルのリレーション・出席率の自動計算を加えた、より実践的なCRUDを実装します。
次回 : Python × Dify × RAGで学ぶ業務システム開発入門【第6回 研修管理編】