从最后提交的检查点恢复导入

目标

把源文件指纹与检查点结合起来，防止错误地接着处理另一个文件。

为什么重要

一个正在导入几千行数据的任务中途崩溃了。运维人员换了文件，用同一个任务 id 重新运行，结果前半部分来自旧文件，后半部分来自新文件。只保存已处理的行号，就无法确认输入的身份。必须把源文件字节的指纹和最后提交的位置一起保存。

步骤

1. 在 /root/work/idem-batch-checkpoint-lab/service.py 中，parse_rows(raw) 读取 JSON 数组的 bytes。每一项都要有 id（非空 str）和 value（不含 bool 的 int），并且禁止 id 重复。违反时抛出 ValueError。返回只包含 {id,value} 的行列表。

先做一次准备。已有的文件不会被覆盖。

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

1. 在 /root/work/idem-batch-checkpoint-lab/service.py 中，source_digest(raw) 是对 bytes 应用 SHA-256 得到的 hex 字符串。不对 JSON 做规范化。

2. 在 /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))。

3. 在 /root/work/idem-batch-checkpoint-lab/service.py 中，begin(path,batch,digest) 对新任务会保存 next_index=0 并返回 0；已有的任务如果指纹相同，就返回 next_index；指纹不同则抛出 ValueError。

4. 在 /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) 并返回。

5. 在 /root/work/idem-batch-checkpoint-lab/service.py 中，checkpoint(path,batch) 返回 next_index，任务不存在时返回 None。

6. 在 /root/work/idem-batch-checkpoint-lab/service.py 中，values(path,batch) 按 id 升序返回该 batch 的 (id,value) 元组。

7. 在 /root/work/idem-batch-checkpoint-lab/service.py 中，import_all(path,batch,raw,size=2,fault=lambda index:None) 校验 size 是（不含 bool 的）正 int，并使用 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} 的行列表。

先做一次准备。已有的文件不会被覆盖。

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) 并返回。

如果每一行都单独提交，检查点与行的状态就会不一致。

保存后用 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) 按 id 升序返回该 batch 的 (id,value) 元组。

把 batch 设为条件，避免其他任务中相同 id 的行混进结果。

保存后用 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) 校验 size 是（不含 bool 的）正 int，并使用 parse_rows、source_digest 和 begin。把剩余的行按每 size 行一批交给 apply_chunk 处理，并返回最终的 checkpoint。

要确认这样的场景：保留第一个批次的成功结果，在第二个批次失败之后恢复执行。

保存后用 bash /opt/lab/checks/idem-batch-checkpoint-lab/08-contract.sh 确认。