Python 3.12+:小批量处理列表时,`itertools.batched()` 挺省事

第 2 页
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 搭一版。