第156回ではCSVの品質ゲートを作り、第157回ではその判定を回帰テストしました。品質ゲートで現在の仕様を固定すると、正当な列名変更や列追加も不一致として検出されます。
検査を緩めて新旧のCSVを混在させるのではなく、旧仕様から新仕様へ変換する工程を分けます。今回はCSVにスキーマバージョンを持たせ、バージョン1だけを受け付けてバージョン2を別ファイルへ出力するスクリプトを、Python標準ライブラリだけで作ります。
今回のゴール
- CSVの新旧仕様をバージョン番号で識別する
- 列名変更と列追加を決めた規則どおりに実行する
- 旧CSVから判断できない値を推測で補わない
- 主キーの欠落や重複を移行前に検出する
- 元ファイルと既存の出力ファイルを上書きしない
- 終了コード、成功時のJSON、出力ファイルを確認して後続処理の実行可否を判断する
新旧の仕様と移行規則を決める
例として、顧客CSVをバージョン1からバージョン2へ変更します。
| バージョン1 | バージョン2 | 移行方法 |
|---|---|---|
| schema_version | schema_version | 値を1から2へ変更 |
| customer_id | customer_id | 値を保持 |
| name | customer_name | 列名だけを変更 |
| 空欄を含めて値を保持 | ||
| なし | status | unknownを設定 |
statusへ一律にactiveを設定することはできません。旧CSVだけでは、顧客が現在も有効か判断できないためです。この例では保留値のunknownを設定し、業務担当者による確認を別工程にします。
移行元CSVを用意する
次の内容をUTF-8のcustomers_v1.csvとして保存します。
schema_version,customer_id,name,email
1,C001,顧客A,contact-a@example.com
1,C002,顧客B,
この実装では、ヘッダーの名前と順序を契約の一部として扱います。列の不足、未承認の追加、並べ替えはすべてエラーになります。
一方向の移行スクリプトを実装する
次のコードをmigrate_customers.pyとして保存します。csv.DictReaderで列数の過不足を調べ、strict=Trueで一部の不正なCSV構文も検出します。これらの挙動はPythonのcsv公式ドキュメントで確認できます。
#!/usr/bin/env python3
import argparse
import csv
import json
import sys
from pathlib import Path
V1_FIELDS = ["schema_version", "customer_id", "name", "email"]
V2_FIELDS = [
"schema_version",
"customer_id",
"customer_name",
"email",
"status",
]
class MigrationError(Exception):
pass
def load_v1(path):
with path.open("r", encoding="utf-8-sig", newline="") as file:
reader = csv.DictReader(file, strict=True)
actual_fields = reader.fieldnames
if actual_fields != V1_FIELDS:
raise MigrationError(
f"v1のヘッダーではありません。"
f"expected={V1_FIELDS}, actual={actual_fields}"
)
rows = []
seen_ids = set()
for record_no, row in enumerate(reader, start=1):
if None in row or any(value is None for value in row.values()):
raise MigrationError(
f"データレコード{record_no}: 列数がヘッダーと一致しません"
)
if row["schema_version"] != "1":
raise MigrationError(
f"データレコード{record_no}: "
f"schema_versionは1である必要があります"
)
customer_id = row["customer_id"]
if not customer_id.strip():
raise MigrationError(
f"データレコード{record_no}: customer_idが空です"
)
if customer_id != customer_id.strip():
raise MigrationError(
f"データレコード{record_no}: "
f"customer_idの前後に空白があります"
)
if customer_id in seen_ids:
raise MigrationError(
f"データレコード{record_no}: "
f"customer_idが重複しています: {customer_id}"
)
if not row["name"].strip():
raise MigrationError(
f"データレコード{record_no}: nameが空です"
)
seen_ids.add(customer_id)
rows.append(row)
if not rows:
raise MigrationError("データ行がありません")
return rows
def migrate_to_v2(rows):
migrated = []
for row in rows:
migrated.append(
{
"schema_version": "2",
"customer_id": row["customer_id"],
"customer_name": row["name"],
"email": row["email"],
"status": "unknown",
}
)
return migrated
def validate_v2(rows):
seen_ids = set()
for record_no, row in enumerate(rows, start=1):
if list(row) != V2_FIELDS:
raise MigrationError(
f"データレコード{record_no}: v2の列構成が不正です"
)
if row["schema_version"] != "2":
raise MigrationError(
f"データレコード{record_no}: v2のバージョンが不正です"
)
customer_id = row["customer_id"]
if not customer_id or customer_id != customer_id.strip():
raise MigrationError(
f"データレコード{record_no}: customer_idが不正です"
)
if customer_id in seen_ids:
raise MigrationError(
f"データレコード{record_no}: "
f"customer_idが重複しています: {customer_id}"
)
if not row["customer_name"].strip():
raise MigrationError(
f"データレコード{record_no}: customer_nameが空です"
)
if row["status"] not in {"active", "inactive", "unknown"}:
raise MigrationError(
f"データレコード{record_no}: statusが不正です"
)
seen_ids.add(customer_id)
def write_v2(path, rows):
try:
with path.open("x", encoding="utf-8", newline="") as file:
writer = csv.DictWriter(
file,
fieldnames=V2_FIELDS,
extrasaction="raise",
)
writer.writeheader()
writer.writerows(rows)
except FileExistsError as exc:
raise MigrationError(
f"出力先が既に存在します: {path}"
) from exc
def main():
parser = argparse.ArgumentParser(
description="顧客CSVをスキーマv1からv2へ移行します"
)
parser.add_argument("source", type=Path, help="v1の入力CSV")
parser.add_argument("output", type=Path, help="v2の出力CSV")
args = parser.parse_args()
try:
if args.source.resolve() == args.output.resolve():
raise MigrationError("入力と出力に同じパスは指定できません")
source_rows = load_v1(args.source)
migrated_rows = migrate_to_v2(source_rows)
validate_v2(migrated_rows)
write_v2(args.output, migrated_rows)
except MigrationError as exc:
print(f"移行失敗: {exc}", file=sys.stderr)
return 3
except (OSError, UnicodeError, csv.Error) as exc:
print(f"処理エラー: {exc}", file=sys.stderr)
return 4
result = {
"status": "ok",
"source_version": "1",
"target_version": "2",
"rows": len(migrated_rows),
"output": str(args.output),
}
print(json.dumps(result, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())
実行結果を確認する
Python 3で、入力とは異なる出力先を指定します。環境によってPython 3の実行コマンド名が異なる場合は、その環境のコマンドに読み替えてください。
python3 migrate_customers.py customers_v1.csv customers_v2.csv
移行と書き込みが完了すると、標準出力に次の1行が出ます。
{"status": "ok", "source_version": "1", "target_version": "2", "rows": 2, "output": "customers_v2.csv"}
生成されるcustomers_v2.csvの内容は次のとおりです。
schema_version,customer_id,customer_name,email,status
2,C001,顧客A,contact-a@example.com,unknown
2,C002,顧客B,,unknown
入力全体と変換結果を検証してから出力を開きます。入力契約に違反している場合、出力ファイルは作られません。出力には排他的な作成モードxを使うため、同名のファイルが存在する場合も上書きせずに停止します。
終了コードと成功時の出力を確認する
| 終了コード | 意味 | 対応 |
|---|---|---|
| 0 | 通常実行では移行とv2検証が完了。ただし、--helpも移行せずに0を返す |
成功時のJSONと出力ファイルも確認する |
| 2 | 引数の不足や未知の引数 | 実行コマンドを修正する |
| 3 | 入力契約違反、同一パス、出力先の既存 | CSVまたは指定先を確認する |
| 4 | ファイル、文字コード、CSV解析のエラー | 入出力環境を確認する |
終了コード2は、argparseが不正な引数に対して返す既定値です。一方、既定の--helpはヘルプを表示して終了コード0を返しますが、移行処理は実行しません。詳細はPythonのargparse公式ドキュメントで確認できます。
同じ出力先を指定して再実行すると、既存ファイルを保護して終了コード3を返し、次のメッセージを標準エラー出力へ出します。
移行失敗: 出力先が既に存在します: customers_v2.csv
自動処理では、終了コード0だけで移行成功と判断しません。通常の位置引数を指定して実行したうえで、終了コードが0であること、標準出力がJSONとして解析できること、statusがokであること、source_versionとtarget_versionが想定どおりであること、JSONのoutputが指定した出力先を示すこと、そのファイルが存在することを確認してから後続の品質検査へ進みます。
書き込み中に入出力エラーが起きた場合は、不完全な出力ファイルが残る可能性があります。終了コード4になったときは出力を検査し、不完全なファイルを削除または退避してから再実行します。
業務フローへ組み込む順序
- 新旧の列、必須条件、許容値、既定値を表にして承認する
- 本番データのコピーを使い、別名の出力先で移行する
- 終了コード0に加え、成功時のJSONと指定した出力ファイルを確認する
- バージョン2用の品質ゲートで件数、主キー、列構成を検査する
unknownなど確認が必要な値を業務担当者が処理する- 後続システムがバージョン2を読めることを確認してから入力先を切り替える
- 旧ファイルを社内の保存ルールに従って保管する
AIに列の対応案を作らせる場合でも、保持する値と不明値の扱いはデータの責任者が決めます。承認した規則をPythonコードに固定すれば、同じ入力から同じ結果を作り、後から点検できます。
この実装の制約
- 入力はカンマ区切りのUTF-8またはBOM付きUTF-8に限定しています。
- 各値は文字列として扱い、日付やメールアドレスの形式は検証しません。
- 全行をメモリに保持するため、利用環境のメモリに収まるCSVが対象です。
unknownを許可しない業務では、後続システムへ渡す前に確認と更新が必要です。- 書き込み中の障害に備えた一時ファイル保存や原子的な切り替えは実装していません。
- 表計算ソフトで開く場合のCSV数式インジェクション対策は含みません。必要な場合は、閲覧環境と対象列を決めたうえで別に対策します。
argparseの既定の--helpも終了コード0を返すため、自動処理では終了コードだけを移行成功の証拠にできません。
まとめ
CSVの仕様変更では、品質ゲートを緩めるのではなく、新旧の仕様と移行規則を分けて管理します。このスクリプトはバージョン1だけを受け付け、判断できない値をunknownとして残し、検証したバージョン2を別ファイルへ出力します。
実データのコピーで移行を試し、終了コード、成功時のJSON、出力ファイルを確認してから、生成結果をバージョン2用の品質ゲートへ渡してください。移行、検査、業務判断を分けることで、未確認の値を本番データへ混入させずに仕様を切り替えられます。