分块迭代，包括生成器

如何解决分块迭代，包括生成器

我有这个用于分块迭代的解决方案：

def chunks(items,chunk_size):
    def get_chunk():
        try:
            for _ in range(chunk_size):
                yield next(iterator)
        except stopiteration:
            return False

    iterator = iter(items)
    while chunk := list(get_chunk()):
        yield chunk


for c in chunks([1,2,3,4,5,6,7,8],3):
    print(c)

它运行良好，与我在 SO 上找到的其他一些解决方案不同，它还处理“无限”生成器，例如：

def natural_numbers():
    n = 0
    while True:
        yield (n := n + 1)


tens = chunks(natural_numbers(),10)
for _ in range(5):
    print(next(tens))

但是，我无法摆脱这种感觉，即应该可以在不调用内部函数的情况下做到这一点。当然，您可以定义一个外部函数并传入 chunk_size 和 iterator，这样可以避免在每次调用 get_chunk() 时重新定义 chunks。但它仍然会有为每个块调用该函数的开销。

有没有人有避免函数调用的建议，但仍然适用于无法索引或切片的可迭代对象？

我使用该函数的主要原因是能够捕获 stopiteration，我认为在生成器推导式中无法在不丢失异常前的最后几项的情况下完成此操作，但也许我'我错了。

解决方法

使用while循环：

def chunks(items,chunk_size):
    iterator = iter(items)
    done = False
    while not done:
        chunk = []
        for _ in range(chunk_size):
            try:
                chunk.append(next(iterator))
            except StopIteration:
                done = True
                break
        if chunk:
            yield chunk

使用 for 循环：

def chunks(items,chunk_size):
    iterator = iter(items)
    chunk = []
    for element in iterator:
        chunk.append(element)
        if len(chunk) == chunk_size:
            yield chunk
            chunk = []
    if chunk:
        yield chunk

保留您的原始想法但删除嵌套函数：

from itertools import islice

def chunks(items,chunk_size):
    iterator = iter(items)
    while chunk := list(islice(iterator,chunk_size)):
        yield chunk

使用第三方库：

>>> from more_itertools import chunked
>>> list(chunked([1,2,3,4,5,6,7,8],3))
[[1,3],[4,6],[7,8]]

,

如果我没记错的话，more-itertools 有 chunked 和 ichunked。

https://more-itertools.readthedocs.io/en/stable/api.html#more_itertools.chunked