L09 ch02 — JSON ファイル永続化

ch02-json 演習解説

⚠️ 自力でドリルを解いてから読んでください

ドリル 1: json.load / json.dump - ファイル読み書き — 模範解答

ドリル名: 01-fill-load-save

import json
from pathlib import Path
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
DB_FILE = Path("tasks.json")

class Task(BaseModel):
    title: str
    done: bool = False

def load() -> list[dict]:
    if not DB_FILE.exists():
        return []
    return json.loads(DB_FILE.read_text(encoding="utf-8"))

def save(tasks: list[dict]) -> None:
    DB_FILE.write_text(
        json.dumps(tasks, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )

@app.get("/tasks")
def get_tasks():
    return load()

@app.post("/tasks", status_code=201)
def create_task(task: Task):
    tasks = load()
    new_task = {"id": len(tasks) + 1, **task.model_dump()}
    tasks.append(new_task)
    save(tasks)
    return new_task

ドリル 1: json.load / json.dump - ファイル読み書き — 解説

ドリル名: 01-fill-load-save

def load() -> list[dict]:
    if not DB_FILE.exists():
        return []
    return json.loads(DB_FILE.read_text(encoding="utf-8"))

def save(tasks: list[dict]) -> None:
    DB_FILE.write_text(
        json.dumps(tasks, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )

> ポイント: loads / dumps の末尾の s は string の s。

> - loads — load from string（文字列を受け取ってPythonオブジェクトを返す）

> - dumps — dump to string（PythonオブジェクトをJSON文字列に変換する）

>

> ファイルから読むときは read_text() で文字列にしてから loads、

> ファイルに書くときは dumps で文字列にしてから write_text() に渡す、

> という2段構えを覚えておくと混乱しない。

ドリル 2: json.dump - JSON ファイルへの保存 — 模範解答

ドリル名: 02-add-save

import json
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

DB_FILE = Path("tasks.json")

def load() -> list[dict]:
    if not DB_FILE.exists():
        return []
    return json.loads(DB_FILE.read_text(encoding="utf-8"))

def save(tasks: list[dict]) -> None:
    DB_FILE.write_text(json.dumps(tasks, ensure_ascii=False, indent=2), encoding="utf-8")

@app.get("/tasks")
def get_tasks():
    return load()

@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    for task in load():
        if task["id"] == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

@app.post("/tasks", status_code=201)
async def create_task(request: Request):
    body = await request.json()
    tasks = load()
    new_task = {"id": len(tasks) + 1, "title": body["title"], "done": body.get("done", False)}
    tasks.append(new_task)
    save(tasks)
    return new_task

@app.patch("/tasks/{task_id}")
async def update_task(task_id: int, request: Request):
    body = await request.json()
    tasks = load()
    for t in tasks:
        if t["id"] == task_id:
            t.update({"title": body["title"], "done": body["done"]})
            save(tasks)
            return t
    raise HTTPException(status_code=404, detail="Task not found")

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    tasks = [t for t in load() if t["id"] != task_id]
    save(tasks)

ドリル 2: json.dump - JSON ファイルへの保存 — 解説

ドリル名: 02-add-save

import json
from pathlib import Path
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
DB_FILE = Path("tasks.json")

class Task(BaseModel):
    title: str
    done: bool = False

def load() -> list[dict]:
    if not DB_FILE.exists():
        return []
    return json.loads(DB_FILE.read_text(encoding="utf-8"))

def save(tasks: list[dict]) -> None:
    DB_FILE.write_text(
        json.dumps(tasks, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )

@app.get("/tasks")
def get_tasks():
    return load()

@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    tasks = load()
    for task in tasks:
        if task["id"] == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

@app.post("/tasks", status_code=201)
def create_task(task: Task):
    tasks = load()
    new_task = {"id": len(tasks) + 1, **task.model_dump()}
    tasks.append(new_task)
    save(tasks)
    return new_task

@app.patch("/tasks/{task_id}")
def update_task(task_id: int, task: Task):
    tasks = load()
    for t in tasks:
        if t["id"] == task_id:
            t.update(task.model_dump())
            save(tasks)
            return t
    raise HTTPException(status_code=404, detail="Task not found")

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    tasks = load()
    tasks = [t for t in tasks if t["id"] != task_id]
    save(tasks)

> ポイント: 変更のたびに load → 変更 → save の3ステップを踏む。

> リクエストをまたいでメモリ上の tasks を使い回さないことが永続化の核心。

>

> 補足: global next_id が不要になった理由は、ID を len(tasks) + 1 で採番するようにしたため。

> ただしタスクを削除すると ID が重複する可能性があるため、本格的なアプリでは UUID や

> データベースのオートインクリメントを使うのが一般的。

ドリル 3: JSON 読み込み + クエリパラメータ - ファイル経由のフィルタ — 模範解答

ドリル名: 03-add-query-filter

import json
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

DB_FILE = Path("tasks.json")

def load() -> list[dict]:
    if not DB_FILE.exists():
        return []
    return json.loads(DB_FILE.read_text(encoding="utf-8"))

def save(tasks: list[dict]) -> None:
    DB_FILE.write_text(json.dumps(tasks, ensure_ascii=False, indent=2), encoding="utf-8")

@app.get("/tasks")
def get_tasks(done: str | None = None):
    tasks = load()
    if done is None:
        return tasks
    if done not in ("true", "false"):
        raise HTTPException(status_code=400, detail="done must be 'true' or 'false'")
    done_value = done == "true"
    return [t for t in tasks if t["done"] == done_value]

@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    for task in load():
        if task["id"] == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

@app.post("/tasks", status_code=201)
async def create_task(request: Request):
    body = await request.json()
    tasks = load()
    new_task = {"id": len(tasks) + 1, "title": body["title"], "done": body.get("done", False)}
    tasks.append(new_task)
    save(tasks)
    return new_task

@app.patch("/tasks/{task_id}")
async def update_task(task_id: int, request: Request):
    body = await request.json()
    tasks = load()
    for t in tasks:
        if t["id"] == task_id:
            t.update({"title": body["title"], "done": body["done"]})
            save(tasks)
            return t
    raise HTTPException(status_code=404, detail="Task not found")

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    tasks = [t for t in load() if t["id"] != task_id]
    save(tasks)

ドリル 3: JSON 読み込み + クエリパラメータ - ファイル経由のフィルタ — 解説

ドリル名: 03-add-query-filter

@app.get("/tasks")
def get_tasks(done: str | None = None):
    tasks = load()
    if done is None:
        return tasks
    if done not in ("true", "false"):
        raise HTTPException(status_code=400, detail="done must be 'true' or 'false'")
    done_value = done == "true"
    return [t for t in tasks if t["done"] == done_value]

ポイント

クエリパラメータの宣言 — def get_tasks(done: str | None = None): と書くだけで FastAPI がURLの ?done=... を受け取る。パスパラメータと違い {} は不要。

done は文字列で届く — URLのクエリ文字列はすべて str として渡される。bool への暗黙変換はない。done == "true" で bool に変換する。

バリデーション位置 — if done not in ("true", "false"): を絞り込み前に置く。None チェックを先に済ませておかないと None not in (...) が True になってしまう。

別解

# bool 型のクエリパラメータを FastAPI に任せる方法
@app.get("/tasks")
def get_tasks(done: bool | None = None):
    tasks = load()
    if done is None:
        return tasks
    return [t for t in tasks if t["done"] == done]

done: bool | None = None と書くと FastAPI が "true"/"false" → True/False に自動変換し、それ以外は 422 を返す。"invalid" のとき 400 ではなく 422 になるが動作は同じ。どちらの実装を選ぶかは要件次第。