最後に確定した位置から取り込みを再開する

目標

原本のフィンガープリントとチェックポイントを組み合わせて、別のファイルで誤って引き継ぐことを防ぎます。

なぜ重要なのか

数千行をインポートしていた作業が、途中で死にました。運用者がファイルを差し替えて、同じ作業idでもう一度実行したところ、前半は古いファイル、後半は新しいファイルという結果ができました。処理した行番号だけを保存していると、入力の同一性を確認できません。原本のバイト列のフィンガープリントと、最後にコミットした位置を、一緒に保持する必要があります。

ステップ

/root/work/idem-batch-checkpoint-lab/service.pyで、parse_rows(raw)は、JSON配列のbytesを読みます。各項目は、id(空でないstr)とvalue(boolを除くint)を持ち、idの重複は禁止します。違反はValueErrorです。{id,value}だけを持つ行のリストを返します。

最初に1回だけ準備してください。既存のファイルは上書きしません。

mkdir -p /root/work/idem-batch-checkpoint-lab
test -e /root/work/idem-batch-checkpoint-lab/service.py || cp /opt/fixtures/ten_labs/idem-batch-checkpoint-lab/service.py /root/work/idem-batch-checkpoint-lab/service.py
cd /root/work/idem-batch-checkpoint-lab

/root/work/idem-batch-checkpoint-lab/service.pyで、source_digest(raw)は、bytesにSHA-256を適用したhex文字列です。JSONを正規化しません。
/root/work/idem-batch-checkpoint-lab/service.pyで、init_db(path)は、imports(id TEXT PRIMARY KEY,digest TEXT NOT NULL,next_index INTEGER NOT NULL)とitems(batch TEXT NOT NULL,id TEXT NOT NULL,value INTEGER NOT NULL,PRIMARY KEY(batch,id))を、冪等に作成します。
/root/work/idem-batch-checkpoint-lab/service.pyで、begin(path,batch,digest)は、新しい作業ならnext_index=0を保存して0、既存の同じフィンガープリントならnext_index、別のフィンガープリントならValueErrorです。
/root/work/idem-batch-checkpoint-lab/service.pyで、apply_chunk(path,batch,start,rows,fault=lambda index:None)は、現在のnext_index==startのときだけ実行し、そうでなければValueErrorです。rowsを順番にitemsに入れ、各挿入のあとにfault(全体のインデックス)を呼びます。すべて成功したら、next_index=start+len(rows)を保存して返します。
/root/work/idem-batch-checkpoint-lab/service.pyで、checkpoint(path,batch)は、next_index、ない作業はNoneです。
/root/work/idem-batch-checkpoint-lab/service.pyで、values(path,batch)は、そのbatchの(id,value)タプルを、id昇順で返します。
/root/work/idem-batch-checkpoint-lab/service.pyで、import_all(path,batch,raw,size=2,fault=lambda index:None)は、boolを除く正のintのsizeを検証し、parse_rows・source_digest・beginを使います。残りの行をsizeずつapply_chunkで処理し、最終的なcheckpointを返します。

参考

インターネットやパッケージのインストールなしで、既存のlab-dev環境で行います。
各ステップは、45秒の採点予算の中で実行されます。実際のsleepやネットワーク呼び出しを追加しないでください。
採点は、提出されたモジュールを新しく読み込み、独立した入力と一時DBで検査します。期待値を定数として返す代わりに、契約を実装してください。
FastAPI公式ドキュメント・pytest公式ドキュメント・Python sqlite3
限界: 入力全体をメモリに読み込む、小さなデータ向けのラボです。大容量ファイルをストリーミングでパースするエンジンだと誇張しません。原本の空白だけが違っても、バイトのフィンガープリントが変わるので、再開を拒否するという、保守的な契約です。外部APIの副作用は、このDBトランザクションには入りません。

入力行の契約を確認する

/root/work/idem-batch-checkpoint-lab/service.pyで、parse_rows(raw)は、JSON配列のbytesを読みます。各項目は、id(空でないstr)とvalue(boolを除くint)を持ち、idの重複は禁止します。違反はValueErrorです。{id,value}だけを持つ行のリストを返します。

最初に1回だけ準備してください。既存のファイルは上書きしません。

mkdir -p /root/work/idem-batch-checkpoint-lab
test -e /root/work/idem-batch-checkpoint-lab/service.py || cp /opt/fixtures/ten_labs/idem-batch-checkpoint-lab/service.py /root/work/idem-batch-checkpoint-lab/service.py
cd /root/work/idem-batch-checkpoint-lab

重複したidを、最後の行で黙って上書きすると、インポートの結果を予測できません。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/01-contract.shで確認してください。

原本のバイト列のフィンガープリントを固定する

/root/work/idem-batch-checkpoint-lab/service.pyで、source_digest(raw)は、bytesにSHA-256を適用したhex文字列です。JSONを正規化しません。

再開は、同じ原本に対してだけ許可する契約です。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/02-contract.shで確認してください。

入力とチェックポイントを保存する

/root/work/idem-batch-checkpoint-lab/service.pyで、init_db(path)は、imports(id TEXT PRIMARY KEY,digest TEXT NOT NULL,next_index INTEGER NOT NULL)とitems(batch TEXT NOT NULL,id TEXT NOT NULL,value INTEGER NOT NULL,PRIMARY KEY(batch,id))を、冪等に作成します。

別々のインポート作業の同じ行idは、分離して保存します。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/03-contract.shで確認してください。

別の原本で再開できないようにする

/root/work/idem-batch-checkpoint-lab/service.pyで、begin(path,batch,digest)は、新しい作業ならnext_index=0を保存して0、既存の同じフィンガープリントならnext_index、別のフィンガープリントならValueErrorです。

作業idと行番号が同じでも、入力ファイルは違うことがあります。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/04-contract.shで確認してください。

バッチと位置をアトミックにコミットする

/root/work/idem-batch-checkpoint-lab/service.pyで、apply_chunk(path,batch,start,rows,fault=lambda index:None)は、現在のnext_index==startのときだけ実行し、そうでなければValueErrorです。rowsを順番にitemsに入れ、各挿入のあとにfault(全体のインデックス)を呼びます。すべて成功したら、next_index=start+len(rows)を保存して返します。

行1つごとに別々にコミットすると、チェックポイントと行の状態がずれます。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/05-contract.shで確認してください。

現在の位置を取得する

/root/work/idem-batch-checkpoint-lab/service.pyで、checkpoint(path,batch)は、next_index、ない作業はNoneです。

最後に処理を試みた位置ではなく、最後にコミットされた位置を読みます。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/06-contract.shで確認してください。

作業ごとの結果を分離する

/root/work/idem-batch-checkpoint-lab/service.pyで、values(path,batch)は、そのbatchの(id,value)タプルを、id昇順で返します。

別の作業の同じidの行が結果に混ざらないように、batchを条件に置きます。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/07-contract.shで確認してください。

途中の失敗のあと、安全に続ける

/root/work/idem-batch-checkpoint-lab/service.pyで、import_all(path,batch,raw,size=2,fault=lambda index:None)は、boolを除く正のintのsizeを検証し、parse_rows・source_digest・beginを使います。残りの行をsizeずつapply_chunkで処理し、最終的なcheckpointを返します。

最初のバッチの成功を保持したまま、2つ目のバッチの失敗後に再開するシナリオを確認します。

保存したあと、bash /opt/lab/checks/idem-batch-checkpoint-lab/08-contract.shで確認してください。