TL;DR
- AIエージェントやバッチの長時間処理は、途中で落ちても済んだところから再開できるようにしておくと安心です。
- 1件終わるごとに状態をJSONに保存し、保存は「一時ファイル → fsync →
os.replace」で原子的に差し替えます。 -
kill -9で強制終了させ、再実行で残りだけが処理されることを確認しました（Python 3.13.5）。
流れ
コード
resumable.py
#!/usr/bin/env python3
"""長時間タスクをチェックポイントから再開する最小例。"""
import json, os, sys, tempfile, time
CKPT = "state.json"
def load():
try:
with open(CKPT, encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return {"done": []}
def save(state):
# 一時ファイルに書いて fsync → os.replace で原子的に差し替える
fd, tmp = tempfile.mkstemp(dir=".", prefix=".state-")
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False)
f.flush(); os.fsync(f.fileno())
os.replace(tmp, CKPT)
def work(item):
time.sleep(float(os.environ.get("STEP_SEC", "0.2"))) # 実際は AI 呼び出しなど
return f"{item}:ok"
def main(items):
state = load()
for it in items:
if it in state["done"]:
continue # 済みのものは飛ばす（冪等）
print("run", it, work(it), flush=True)
state["done"].append(it)
save(state)
print("all done:", len(state["done"]))
if __name__ == "__main__":
main([f"task{i:02d}" for i in range(10)])
ポイント:
- 済んだタスクは飛ばすので、何度実行しても結果が同じ（冪等）。
-
os.replaceは、成功すれば置き換え先が新しいファイルになる操作です（POSIXでは原子的、Python公式ドキュメント）。書き込み途中で落ちてもstate.jsonが半端な内容になりません。
検証：途中で kill -9
$ python3 resumable.py & p=$!; sleep 0.9; kill -9 $p
run task00 task00:ok
run task01 task01:ok
run task02 task02:ok
run task03 task03:ok
killed rc=137
$ cat state.json
{"done": ["task00", "task01", "task02", "task03"]}
$ python3 resumable.py
run task04 task04:ok
...
run task09 task09:ok
all done: 10
再実行では task04 から始まり、残った一時ファイル（.state-*）は0個でした。
注意
- 「処理した」と「保存した」の間で落ちると、そのタスクはもう一度実行されます。外部に副作用がある処理（投稿・課金など）は、処理側でも重複を防ぐ仕組みを入れてください。
- 件数が多い場合は、JSONの丸ごと書き直しより SQLite などを検討します。