Python講座 / 実習：APIデータを重複させずに保存する
APIの応答全体を検証してから、安定した外部IDを使って保存します。この実習はJSON取得とSQLiteのトランザクションをつなぎ、再取得時は同じIDの名前と金額を更新します。
同一性と再実行の意味を決める
upsertには、同じ外部IDが同じ取引を指し続ける契約が必要です。名前と金額が同じだけでは同一取引とはいえません。同じ応答内にIDが複数あれば、この例は最後の値を使います。別の方針が必要なら書き込み前に重複を拒否します。
全件を検証した後、データベースの書き込みを一つのトランザクションで行います。ただしネットワーク取得とローカル保存が分散トランザクションになるわけではありません。Python 3.12以降を使います。
ローカル応答を用意する
新しいフォルダーへexpenses.jsonを保存します。
[{"id":"e1","name":"coffee","amount":3500},{"id":"e2","name":"lunch","amount":8000}]
そのフォルダーで一つ目のターミナルからサーバーを起動します。
python -m http.server 8000 --bind 127.0.0.1
二つ目のターミナルで取得する間は起動したままにします。ポートを変更したら、取得側のURLにも同じ番号を使います。
収集プログラムの全体を実行する
collect.pyへ保存します。
import json
import sqlite3
import sys
from urllib.request import urlopen
with urlopen(sys.argv[1], timeout=5) as response:
if response.status != 200:
raise ValueError("unexpected status")
raw = response.read(1_000_001)
if len(raw) > 1_000_000:
raise ValueError("response too large")
data = json.loads(raw.decode("utf-8"))
if not isinstance(data, list) or len(data) > 10000:
raise ValueError("invalid batch")
MAX_AMOUNT = 1_000_000_000_000
records = []
for row in data:
if not isinstance(row, dict):
raise ValueError("invalid row")
key, name, amount = row.get("id"), row.get("name"), row.get("amount")
if not isinstance(key, str) or not key.strip() or not isinstance(name, str) or not name.strip():
raise ValueError("invalid identity")
if type(amount) is not int or not 0 <= amount <= MAX_AMOUNT:
raise ValueError("invalid amount")
records.append((key, name, amount))
con = sqlite3.connect("collected.db", autocommit=False)
try:
with con:
con.execute("CREATE TABLE IF NOT EXISTS expenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, amount INTEGER NOT NULL CHECK(amount BETWEEN 0 AND 1000000000000))")
con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
count = con.execute("SELECT COUNT(*) FROM expenses").fetchone()[0]
total = sum(row[0] for row in con.execute("SELECT amount FROM expenses"))
print("stored:", count, "total:", total)
finally:
con.close()
プロジェクトの環境で実行します。
python collect.py http://127.0.0.1:8000/expenses.json
stored: 2 total: 11500が予想結果です。もう一度実行しても4件には増えず、2件のままです。失敗時はトレースバックと終了状態を確認してから後続処理を判断します。
更新・失敗・収集の完全性を試す
同じ応答を二回処理して件数と合計を比較します。e2を9000へ変えると2件、合計12500になります。金額を文字列へ変えたときは拒否され、以前の保存内容が残ることを確認します。最後に元の応答へ戻し、Ctrl+Cでサーバーを止めます。
外部サービスでは認証、ページ分割、レート制限、変更・削除された記録の扱いも必要です。数ページの成功だけで収集全体の完了とは限りません。エラーをすべて空リストに変えたり、無制限に再試行したりすると欠落を隠します。HTTPクライアントとSQLiteで、それぞれの失敗とトランザクションの境界を確認できます。
保存前に各金額を0〜1,000,000,000,000の整数に制限します。各行が64ビット内でもSQLiteの整数SUMは合計でオーバーフローするため、この例は読み出した整数をPythonのsumで正確に合計します。金額を浮動小数点のtotal()に置き換えません。既存の制約は自動更新されないので、新しいフォルダーか別途移行を使います。初回は2行・11,500、同じ応答の再実行も同じ、e2を9,000にすると2行・12,500です。不正値は書き込み前に拒否して以前のDBを保ちます。
動作と適用条件の公式説明：Python: urllib.request.urlopen · SQLite: sum and total
