中文
devexamples

Batch an Iterator Lazily

Consume an iterable in fixed-size batches with a generator, processing large streams without loading everything into memory.

PythonData Integrity

Code

snippet
from typing import Iterable, Iterator, List, TypeVar

T = TypeVar("T")


def batched(items: Iterable[T], size: int) -> Iterator[List[T]]:
    if size <= 0:
        raise ValueError("size must be positive")
    batch: List[T] = []
    for item in items:
        batch.append(item)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch


for group in batched(range(7), 3):
    print(group)

Related Examples