Split a Stream into Batches with batched

An API accepts at most 100 records per request, and your records come from a generator that reads a large file. Slicing records[i:i + 100] needs a list. itertools.batched works on any iterable, generators included.

from itertools import batched

records = iter(["a", "b", "c", "d", "e"])
assert list(batched(records, 2)) == [("a", "b"), ("c", "d"), ("e",)]

batched(iterable, n) yields tuples of n items, in input order. The last tuple holds whatever remains, so it can be shorter. Each step reads only enough input to fill the next batch. Empty input gives no batches, and an n below 1 raises ValueError.

Because the last batch can be short, do not unpack every batch into a fixed number of names. On an odd number of values, for a, b in batched(values, 2) raises ValueError at the last batch.

Convert each batch as you go

If the API wants lists, convert each tuple as it comes:

from collections.abc import Iterable, Iterator
from itertools import batched


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


assert list(list_pages(["a", "b", "c"], 2)) == [["a", "b"], ["c"]]

pages = list_pages(["a"], 0)  # no error yet
try:
    next(pages)
except ValueError:
    print("the error appears at the first next(), not at the call")

list(batch) copies one batch, which takes time in proportion to the batch size. It does not copy the records themselves. Because list_pages is a generator function, it holds one batch at a time. Collecting all the pages into one list would hold the whole input in memory again.

Check the argument when the function is called

"Generators and One-Shot Iterators" showed that calling a generator function runs none of its body. So list_pages(["a"], 0) above does not raise. The ValueError from batched appears only when the caller starts the loop, which may be far from the line that passed the bad size.

To reject a bad size at the call, write a plain function that checks the argument and then returns a generator:

from collections.abc import Iterable, Iterator
from itertools import batched


def eager_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()


try:
    eager_pages(["a"], 0)
except ValueError:
    print("the error appears at the call")

eager_pages has no yield of its own, so it is a plain function, and its check runs during the call. The inner generator function pages still waits for the first next() before it reads any records. The return type is the same Iterator[list[str]], but the error comes at a different time. Callers that catch it may need to move their try block.

Check your understanding Does checking the size early guarantee that the loop will succeed? Show answer

No. The records can still raise an error while the loop reads them, and so can the code that uses each page. The early check covers only the size rule. It does not read the records ahead of time or hide their errors.

Check your understanding Why is zip(*[iter(records)] * size) not the same as batched? Show answer

That expression passes one iterator to zip size times, so each tuple takes the next size records. But zip stops as soon as it cannot fill a whole tuple, so it drops a short last batch. batched keeps that batch. Use the zip version only when dropping the tail is what you want, and say so in the code.

Going further

Python 3.13 added batched(iterable, n, strict=True), which raises ValueError when the last batch is short. The error comes during the loop, after your code has already used the full batches, so it does not undo any work. Python 3.12 has no strict argument.