Ryan小满Lv1#10 顺带一提,这种分批方式拿来做失败重试也挺顺。哪一批出问题,就只重跑那一批,不用整份列表再切一遍。
顺手加个批次号会更好排查:
from itertools import batched
for i, chunk in enumerate(batched(rows, 500), 1):
print(i, len(tuple(chunk)))
哪一批出问题,日志里一眼就能看到。
顺手补一个边界:batched() 的 n 不能小于 1,参数如果是从配置里读的,最好先兜一下,不然会直接报错。
这种小坑挺容易漏,尤其是批大小本来就可能被人改成 0 的时候。
还有个我挺喜欢的点:它出来是 tuple,拿去做并发分发时反而省心一点,基本不用担心这一批被后面的代码顺手改掉。
像先分批,再丢线程池/进程池,这种链路就很顺。
我这边更常把它和限流绑一起用。比如外部接口一次只收 100 条,batched() 直接把节奏卡住了,代码比自己手动记计数器清爽。
补个容易混的点:batched() 不会帮你补齐最后一批。
如果后面接口必须固定长度,就得自己补一下;只是想分组处理的话,它这个行为正好。
这个点我会顺手标一下版本边界:batched 是 3.12 才有,旧环境会直接 ImportError。
要兼容的话,我一般就先留个 more-itertools.chunked 兜底,或者自己用 itertools.islice 搭一版。