中文
devexamples

Batch an Iterator into Fixed-Size Chunks

Group any iterable into fixed-size lists lazily, preserve the final partial batch, and stop cleanly when the source is exhausted.

PythonData Fetching

Code

snippet
from collections.abc import Iterable, Iterator
from itertools import islice
from typing import TypeVar

T = TypeVar("T")


def chunked(items: Iterable[T], size: int) -> Iterator[list[T]]:
    if size < 1:
        raise ValueError("size must be at least 1")

    iterator = iter(items)
    while True:
        batch = list(islice(iterator, size))
        if not batch:
            return
        yield batch


def main() -> None:
    for batch in chunked(range(1, 11), 3):
        print(batch)

    print(list(chunked([], 2)))
    print(list(chunked((value for value in range(3)), 5)))

    try:
        list(chunked(range(5), 0))
    except ValueError as error:
        print("error:", error)


if __name__ == "__main__":
    main()

Related Examples