batched: chunking without the off-by-one

Everybody has written this:

for i in range(0, len(records), size):
    batch = records[i : i + size]
    insert_many(batch)

It is correct, the ragged tail works, and it has exactly one problem: it needs len() and slicing. So it works on a list and breaks on a generator, a database cursor, a file handle, a socket, a paginated API response — which is the entire set of situations where you actually needed to batch, because those are the ones too large to hold in memory.

Since 3.12:

from itertools import batched

for batch in batched(records, size):
    insert_many(batch)

batched(iterable, n, *, strict=False). Works on any iterable, holds one batch at a time, and the ragged final batch comes out shorter with no special case. n < 1 raises ValueError. The strict=True keyword arrived in 3.13 — on 3.12 the parameter does not exist, so code that passes it fails at runtime on the older interpreter.

Batches are tuples, and that is a typing trap

batched yields tuple[T, ...] — a variadic tuple, not a fixed-length one. So this type-checks:

for left, right in batched(values, 2):    # mypy: fine
    ...

and raises ValueError: not enough values to unpack on the final batch whenever len(values) is odd. The checker cannot help, because tuple[int, ...] genuinely might have two elements. The runtime fix is batched(values, 2, strict=True) (3.13+), which raises if the last batch is short — turning a lurking ValueError deep in your loop into an immediate, well-named failure at the source.

Contrast itertools.pairwise, which is typed Iterator[tuple[T, T]] — fixed-length, so unpacking there is arity-checked. Same-looking code, two different guarantees, and the only way to know is to have read the signature.

def bulk_pages(records: Iterable[str], size: int) -> Iterator[list[str]]:
    ...
    for batch in batched(records, size):
        yield list(batch)

Validate eagerly, generate lazily

A generator function defers everything to the first next(), including argument validation:

def bulk_pages(records, size):
    if size < 1:
        raise ValueError("size must be >= 1")   # runs on first next(), not on call
    for batch in batched(records, size):
        yield list(batch)

bulk_pages(records, 0) returns a generator object quite happily. The ValueError fires later — often in a different function, several frames from the bad argument, with a traceback that points at the consumer rather than the caller. In a pipeline assembled in one place and consumed in another, that is a genuinely hard bug to read.

The fix is a plain function that validates and returns an inner generator:

def bulk_pages(records: Iterable[str], size: int) -> Iterator[list[str]]:
    if size < 1:
        raise ValueError("size must be >= 1")

    def pages() -> Iterator[list[str]]:
        for batch in batched(records, size):
            yield list(batch)

    return pages()

The outer function body runs at call time; the inner one stays lazy. The return type is identical, so no caller changes.