Python実践講座 18：総合演習・JSON保存とargparse

第17回のdomain.pyと同じフォルダにtasks.pyを作ります。保存失敗を成功に見せず、破損データを空として扱わない設計を実装します。

import argparse
import json
import os
import sys
import tempfile
from dataclasses import asdict
from pathlib import Path
from domain import Task


def load(path: Path) -> list[Task]:
    try:
        with path.open(encoding="utf-8") as source:
            text = source.read(1_000_001)
    except FileNotFoundError:
        return []
    if len(text) > 1_000_000:
        raise ValueError("保存ファイルが大きすぎます")
    data = json.loads(text)
    if (not isinstance(data, dict) or set(data) != {"version", "tasks"}
            or type(data["version"]) is not int or data["version"] != 1
            or not isinstance(data["tasks"], list)):
        raise ValueError("保存形式が不正です")
    tasks = [Task.from_dict(item) for item in data["tasks"]]
    if len({task.id for task in tasks}) != len(tasks):
        raise ValueError("IDが重複しています")
    return tasks


def save(path: Path, tasks: list[Task]) -> None:
    text = json.dumps({"version": 1, "tasks": [asdict(t) for t in tasks]},
                      ensure_ascii=False, indent=2)
    if len(text) > 1_000_000:
        raise ValueError("保存ファイルが大きすぎます")
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8",
                dir=path.resolve().parent, delete=False) as output:
            temporary = Path(output.name)
            output.write(text)
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary, path)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description="学習用タスク管理")
    parser.add_argument("--file", type=Path, default=Path("tasks.json"))
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("list")
    add = commands.add_parser("add")
    add.add_argument("title")
    done = commands.add_parser("done")
    done.add_argument("id", type=int)
    args = parser.parse_args(argv)
    try:
        tasks = load(args.file)
        if args.command == "list":
            for task in tasks:
                mark = "x" if task.done else " "
                print(f"{task.id} [{mark}] {task.title}")
        elif args.command == "add":
            new_id = max((task.id for task in tasks), default=0) + 1
            tasks.append(Task(new_id, args.title))
            save(args.file, tasks)
            print(f"追加: {new_id}")
        else:
            if not any(task.id == args.id for task in tasks):
                raise ValueError("IDがありません")
            tasks = [task.complete() if task.id == args.id else task for task in tasks]
            save(args.file, tasks)
            print(f"完了: {args.id}")
        return 0
    except (OSError, ValueError) as exc:
        print(f"エラー: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

python tasks.py add "Pythonを復習"、python tasks.py done 1、python tasks.py list の順で実行すると、ID1が完了状態で表示されます。保存先を変えるなら python tasks.py --file practice.json list のように、サブコマンドより前に指定します。

一時ファイルを保存先と同じディレクトリに作り、閉じた後で置き換えます。これは同時更新の競合を防ぐロックではありません。またディレクトリの同期まではしていないため、停電を含む完全な耐久性を保証する例でもありません。

練習と解答

練習：tasks.jsonへ不正なJSONを書き、addを実行します。

解答：読み込みでエラーになり、元のファイルを維持します。FileNotFoundErrorだけを空の初期状態とし、権限エラーや破損を同じ扱いにしない点が重要です。

公式資料

argparseとos.replaceを参照できます。

Python全20回の目次 ｜ 前の回 ｜ 次の回