Python itertools.batched 的 strict=True 怎么避免尾批次漏数据：批处理边界与异常验收

批量同步订单时，最容易漏掉的不是中间数据，而是最后那一小撮记录：总数 10 条、每批 4 条，默认的 itertools.batched 会正常产出两批 4 条和一批 2 条。如果下游接口把每个元组都当成完整批次，尾批次就可能悄悄绕过校验。Python 3.13 起可以用 strict=True 把这个边界直接变成异常。

需要每批都恰好有 n 条时使用 strict=True；允许最后一批不足时保留默认值；如果业务要补齐而不是拒绝，就在调用前明确写出补齐规则，别把短批次误当成完整批次。

默认模式为什么会留下一个短尾批次

先看最小复现。这里的 orders 有 10 条，批大小是 4。默认模式会把输入拆成 3 个元组，最后一个只有 2 条；这不是异常，而是函数的设计语义。

from itertools import batched

orders = [f"order-{i:02d}" for i in range(1, 11)]
for batch in batched(orders, 4):
    print(len(batch), batch)

# 4 ('order-01', 'order-02', 'order-03', 'order-04')
# 4 ('order-05', 'order-06', 'order-07', 'order-08')
# 2 ('order-09', 'order-10')

batched 返回的是惰性迭代器，不会为了计算总数而先把所有输入复制一遍。每次凑够 4 条就产出一批，输入耗尽时再产出剩余记录。因此“最后一批短”应当被视为一个需要业务判断的状态，而不是库函数出错。

strict=True 把尾批次边界变成 ValueError

如果下游接口要求每次请求都必须有 4 条订单，可以打开 strict=True。完整批次会照常产出；当迭代器耗尽而当前批次只有 2 条时，函数抛出 ValueError。

from itertools import batched

def send_full_batches(order_ids, batch_size=4):
    for batch in batched(order_ids, batch_size, strict=True):
        send_to_gateway(batch)

def send_to_gateway(batch):
    if len(batch) != 4:
        raise RuntimeError("gateway requires four orders")
    print("sent", batch)

try:
    send_full_batches(["A", "B", "C", "D", "E", "F"], 4)
except ValueError as exc:
    print("reject input:", exc)

# sent ('A', 'B', 'C', 'D')
# reject input: batched(): incomplete batch

这里要注意一个容易被忽略的事实：异常发生在消费最后一批时，前面的完整批次已经被交给 send_to_gateway。如果业务要求“整批输入要么全成功、要么一条都不发送”，就不能只包一层 try/except；应先把批次写入待处理记录，或先做完整性预检，再执行外部副作用。

补齐、拒绝和保留短批次怎么选

三种策略都合理，关键看下游契约。文件分片或消息批量接口通常适合保留短批次；固定长度向量、必须四条一起结算的接口适合拒绝；如果协议允许占位记录，则可以补齐，但补齐值必须和真实订单明确区分。

from itertools import batched

def padded_batches(items, size, fill_value=None):
    for batch in batched(items, size):
        if len(batch) 

不要用 strict=True 之后捕获异常，再把原来的迭代器继续交给别的流程，期待它从头开始。迭代器已经被消费过；更稳的做法是保存原始列表、文件偏移或可重放的查询游标，然后在明确记录失败原因后重新构建输入。

两个参数边界要在测试里写出来

n 小于 1 会立即抛出 ValueError，不应把它当成“空批次”处理。输入为空时不会产出任何批次；输入长度刚好是 n 的整数倍时，strict=True 不会报错。

from itertools import batched

assert list(batched([], 4, strict=True)) == []
assert list(batched([1, 2, 3, 4], 4, strict=True)) == [(1, 2, 3, 4)]

try:
    list(batched([1, 2], 0, strict=True))
except ValueError as exc:
    assert "at least one" in str(exc)

try:
    list(batched([1, 2, 3, 4, 5], 4, strict=True))
except ValueError as exc:
    assert "incomplete batch" in str(exc)

如果项目还要支持 Python 3.12，不能直接传入 strict，因为该关键字参数是 3.13 才加入的。兼容层可以按运行时版本分支，或者继续使用项目自己的分批函数，并为两条路径写同一组尾批次测试。

把批次验收放在副作用之前

实际同步任务里，建议先确定输入是否满足批量契约，再决定是否发送。最小验收清单是：批大小明确、空输入行为明确、尾批次策略明确、异常后能重放输入，以及下游已经收到的完整批次有可追踪记录。

若采用 strict=True，可以把 ValueError 当成输入完整性失败，而不是网络重试信号。这样既不会重复发送前面的完整批次，也不会把短尾批次静默交给一个只接受固定数量的接口。

相关问题

batched 会一次性读取整个列表吗？

不会。它按批次惰性消费输入；但如果输入是列表，列表本身已经在内存中。真正需要控制读取量时，可以把文件行迭代器或生成器直接传入。

strict=True 会把短批次补成完整批次吗？

不会。它只负责发现最终批次不足并抛出 ValueError，补齐需要由业务代码明确实现。

什么时候不该使用 strict？

只要协议允许最后一批不足，例如分页写入、日志分片或文件上传，默认模式往往更自然；把短批次当错误会增加重放和失败处理成本。

判断标准可以压缩成一句话：下游要固定长度就用 strict=True 并准备可重放输入，下游接受尾批次就保留默认值，协议需要占位就显式补齐。不要让一个隐式的短元组替你做业务决策。