PythonでCSVを安全に置き換える ― 書込み途中で失敗しても元データを残す

# 注意: 既存のファイルを直接書き換える例

with open("batch.csv", "w", encoding="utf-8", newline="") as fp:
    fp.write("id,status\n")

    # ここで処理が停止すると、元データは戻りません。

from __future__ import annotations

import csv
import os
import tempfile
from pathlib import Path
from collections.abc import Iterable, Mapping

COLUMNS = ("id", "status")


def replace_csv(target: Path, rows: Iterable[Mapping[str, str]]) -> None:
    """CSVを書き終えるまで元ファイルを変更しない。"""
    target = Path(target)
    temporary: Path | None = None
    try:

        # 置換先と同じフォルダーに一時ファイルを作成する

        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="", delete=False,
            dir=target.parent, prefix=f".{target.name}.", suffix=".tmp"
        ) as fp:
            temporary = Path(fp.name)
            writer = csv.DictWriter(fp, fieldnames=COLUMNS, extrasaction="raise")
            writer.writeheader()

            for row in rows:

                # 入力列の過不足も検出し、失敗したら置換しない

                if set(row) != set(COLUMNS):
                    raise ValueError("CSV row must have exactly id and status")
                writer.writerow(row)

            # Pythonのバッファを出し、OSにファイル内容の同期を依頼

            fp.flush()
            os.fsync(fp.fileno())

        # 全件を書き終えてから、元CSVの名前へ置き換える

        os.replace(temporary, target)
        temporary = None
    finally:

        # 置換前に失敗したら、作った一時ファイルだけを削除

        if temporary is not None:
            temporary.unlink(missing_ok=True)


if __name__ == "__main__":

    # 毎回新しいディレクトリにサンプルCSVを作る

    with tempfile.TemporaryDirectory(prefix="csv-atomic-demo-") as folder:
        csv_path = Path(folder) / "batch.csv"
        csv_path.write_text("id,status\nold,OLD\n", encoding="utf-8")

        replace_csv(csv_path, (
            {"id": "A-001", "status": "DONE"},
            {"id": "A-002", "status": "PENDING"},
        ))

        with csv_path.open(newline="", encoding="utf-8") as fp:
            entries = list(csv.DictReader(fp))
        print(f"[RESULT] rows={len(entries)} ids={[x['id'] for x in entries]}")

python3 atomic_csv.py
python3 -m unittest -v test_atomic_csv.py

[RESULT] rows=2 ids=['A-001', 'A-002']
Ran 3 tests in 0.002s
OK

with tempfile.NamedTemporaryFile(
    mode="w",
    encoding="utf-8",
    newline="",
    delete=False,
    dir=target.parent
) as fp:
    ...

with csv_path.open(encoding="utf-8", newline="") as fp:
    rows = list(csv.DictReader(fp))

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as folder:
    path = Path(folder) / "batch.csv"
    path.write_text("id,status\nOLD,OK\n", encoding="utf-8")

    try:
        replace_csv(path, (
            {"id": "A", "status": "DONE"},
            {"id": "MISSING"},
        ))
    except ValueError as error:
        print(f"[EXPECTED_ERROR] {error}")

    print(path.read_text(encoding="utf-8"))

[EXPECTED_ERROR] CSV row must have exactly id and status
id,status
OLD,OK