chunked filter

Author: rafalkrupinskiCreated Jul 4, 2023Updated Jul 9, 2023

Is boltons a good home for a filter_chunked function?

python

def filter_chunked(iterable: Iterable[T], chunk_size: int, func: Callable[[Iterable[T]], Iterable[bool]]) -> Iterator[T]:
    for chunk in boltons.iterutils.chunked(iterable, chunk_size):
        for item, allow in zip(chunk, func(chunk)):
            if allow:
                yield item

It behaves similar to filter(), except the predicate is called with chunks of input. It's useful the predicate must make calls to external services, but can do it in batches.

Source: mahmoud/boltons

View original on GitHub View discussion on GitHub