如何逐页阅读可迭代的页面？答案

【问题标题】：How to read an iterable page by page?如何逐页阅读可迭代的页面？
【发布时间】：2021-12-30 04:35:55
【问题描述】：

我尝试了很多方法来逐页读取项目，而无需将每个页面加载到列表中并返回，这可能会在大页面上占用太多内存。我想避免仅仅为了必须再次扫描列表以对每个项目进行后处理而获得大量项目列表。

所以要么我得到将不断返回空数据并用无限数量的空列表填充pages 列表的生成器（使用page_from_iterable2 时），或者我只得到第一页（如page_from_iterable1。

任何提示我做错了什么？

谢谢。

from typing import Iterable, Iterator


def read_paginated_items(
    it: Iterator,
    page_size: int,
) -> Iterable:
    for _ in range(page_size):
        try:
            yield next(it)
        except StopIteration:
            return


def page_from_iterable1(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iter(iterable)
    page_items_generator = read_paginated_items(it, page_size)
    yield page_items_generator


def page_from_iterable2(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iter(iterable)
    while page_items_generator := read_paginated_items(it, page_size):
        yield page_items_generator
    

def test_read_by_page():
    pages = []
    for page in page_from_iterable1([1, 2, 3, 4, 5], 2):
        page_items = [item for item in page]
        pages.append(page_items)

    assert pages == [[1, 2], [2, 3], [5]]

【问题讨论】：

标签： python iterator generator

【解决方案1】：

你需要一些方法来维护你的生成器中的状态。

这听起来像是 iterable 类的工作。

from typing import Iterable, Iterator

class Page:
    def __init__(self, it: Iterator, page_size: int):
        self.it = it
        self.page_size = page_size
        self.done = False
        self.item = 0
    
    def __iter__(self):
        self.item = 0
        return self
    
    def __next__(self):
        while self.item < self.page_size:
            try:
                self.item += 1
                return next(self.it)
            except StopIteration:
                # at this point the entirety of the original
                # iterator is consumed
                # self.done is our way of telling the generator
                # to stop yielding the instance of Page
                self.done = True
                raise
        # here we have reached the end of the page so we just reset the
        # item count in __iter__. The entry point on each iteration.
        raise StopIteration


def page_from_iterable(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iter(iterable)
    page = Page(it, page_size)
    while not page.done:
        yield page
    

def test_read_by_page():
    pages = []
    for page in page_from_iterable([1, 2, 3, 4, 5], 2):
        page_items = [item for item in page]
        pages.append(page_items)

    print(pages)

test_read_by_page()

生成器通过yield为每个页面使用相同的迭代器来工作。由于它继续 yielding 相同的 Page 实例，原始迭代器 (self.it) 的状态被保持。

通常在__iter__ 方法中，状态会被重置。但由于我们希望继续从该可迭代对象进行迭代，您只需将项目计数重置回0。

【讨论】：

并非每个可迭代对象都提供或可以提供可用的__length_hint__。这就是为什么它被称为hint。
是的。无论如何，我的偏好倾向于实现迭代器类。我在事后添加了__length_hint__ 部分，因为我觉得应该有一个只涉及生成器的解决方案。但是对于像页面这样的动态生成的内容，我想我可以假设它是不准确的。我会删除它。

【解决方案2】：

如果您愿意用 0 个元素测试生成的分页，可以进行简化：

from typing import Iterable, Iterator
import itertools

def paginate(
    it: Iterator,
    page_size: int,
) -> Iterable:
    try:
        for _ in range(page_size):
            yield it.__next__()
    except StopIteration:
        pass

def page_from_iterable(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iterable.__iter__()
    while True:
        yield paginate(it, page_size)

def test_read_by_page():
    pages = []
    for page in page_from_iterable([1, 2, 3, 4, 5], 2):
        page = list(page)
        if not page:
            break
        pages.append(page)
    print(pages)

test_read_by_page()

打印：

[[1, 2], [3, 4], [5]]

【讨论】：

谢谢。我不喜欢if not page: break 的想法，但是可以使用while 构造来删除它。
那while 将如何工作？ page 是生成器表达式，不能直接检查 0 长度。知道它是否会生成一个空列表的唯一方法是将它转换为一个列表并像我上面所做的那样检查它，或者您的计划是否实际上只是迭代这些生成器，我认为这确实是生成生成器的重点开始是因为你担心列表会占用太多内存，当你检测到你有一个空的生成器时，会跳出循环。
当然，在某些时候我需要从项目中生成一个列表。我只是不想直接向我提供列表，所以我有机会修改项目。例如，我的用例是当每个项目都是一个元组时，我只想要第一个元素。看看我的回答，看看我的意思。 while 不会测试生成器，但一旦我有机会修改它们。

【解决方案3】：

def page_from_iterable2(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iter(iterable)
    while page_items_generator := read_paginated_items(it, page_size):
        yield page_items_generator

这里的问题很简单，page_items_generator 是...一个生成器，而不是生成的项目。每次通过循环，您都会创建一个新的生成器对象； while 条件通过（因为生成器对象是真实的）；你产生了那个对象，实际上没有从嵌套生成器中读取任何内容。

您需要明确收集结果：

def pages_from_iterable(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iter(iterable)
    while page := list(read_paginated_items(it, page_size)):
        yield page

现在，每次循环时，创建的生成器都用于读取最多page_size 个项目，创建一个项目列表。当源项用尽时，您可能会得到一个少于page_size 项的列表，然后是一个空列表（在这两种情况下都是由于StopIteration 的处理。由于空列表是错误的，while 循环中断并且不会产生该列表。

这意味着我们不需要从外部收集每页结果：

def test_read_by_page():
    for page in pages_from_iterable([1, 2, 3, 4, 5], 2):
        print(page)

也许您希望将页面结果的收集推迟到生成器之外。不幸的是，这根本行不通：无论生成器会生成什么，生成器都是真实的，在一般情况下，弄清楚它们会生成什么的唯一方法就是让它们这样做。幸运的是，您的 page 大小是有限的并且可能很小，所以这仍然可以让您避免任何内存问题。毕竟，这就是分页的意义，对吧？

【讨论】：

感谢您的回答。调用list() 确实可以解决我的问题，但会创建一个页面上所有项目的列表，这正是我想要避免的。我希望能够即时迭代它们。
那么恐怕你还需要一些外部逻辑来检测一个空页面并突破。
其实，等等。这可能会被残酷地攻击。我将单独发布，因为它完全是一种不同的方法。

【解决方案4】：

调用 list() 确实可以解决我的问题，但会创建一个包含页面所有项目的列表，这正是我想要避免的。我希望能够即时迭代它们。

__length_hint__ 不能可靠地解决这个问题；但是如果我们允许在页面生成时从每个页面推测性地读取一个项，我们可以：

通过尝试读取一项来测试页面是否为空
如果是，则返回一个哨兵值，而不是由外部生成器适当处理的生成器
否则，请使用包装器将商品放回原处

看起来像：

def generator_with_prepended(iterator, value):
    yield value
    yield from iterator

def sentinelize_empty_generator(generator):
    it = iter(generator)
    try:
        first = next(it)
        return generator_with_prepended(it, first)
    except StopIteration:
        return None # which is falsey

# read_paginated_items as before

def pages_from_iterable(
    iterable: Iterable,
    page_size: int,
) -> Iterable:
    it = iter(iterable)
    while page_items_generator := sentinelize_empty_generator(read_paginated_items(it, page_size)):
        yield page_items_generator

我们再次需要从外部收集结果：

def test_read_by_page():
    for page in pages_from_iterable([1,2,3,4,5], 2):
        for item in page:
            print(item)
        print('---')

【讨论】：

为什么要使用iter()函数？生成器应该是可迭代的，并且在pages_from_iterable 中您使用类型提示。我不是 python 专家，所以我问是否有我看不到的原因。
经过审查，实际上可能没有必要，我只是想小心生成器状态。

【解决方案5】：

感谢大家的帮助，这是我想出的：

from typing import Iterator

import pytest


class PageItems:
    def __init__(
        self,
        iterator: Iterator,
        page_size: int,
    ):
        self.items_generator = self._create_items_generator(iterator, page_size)

    @staticmethod
    def _create_items_generator(
        iterator: Iterator,
        page_size: int,
    ):
        for _ in range(page_size):
            try:
                yield next(iterator)
            except StopIteration:
                return

    def __iter__(self):
        return self

    def __next__(self):
        return next(self.items_generator)


def test_read_one_page():
    iterable = [1, 2, 3, 4, 5]

    page_items = PageItems(iter(iterable), 3)
    assert next(page_items) == 1
    assert next(page_items) == 2
    assert next(page_items) == 3

    with pytest.raises(StopIteration):
        next(page_items)


def test_read_pages():
    iterable = [1, 2, 3, 4, 5]
    pages = []
    iterator = iter(iterable)

    while page_items := list(PageItems(iterator, 2)):
        pages.append(page_items)

    assert pages == [[1, 2], [3, 4], [5]]

def test_read_pages_modified_items():
    iterable = [(1, "A"), (2, "B"), (3, "C"), (4, "D"), (5, "E")]
    pages = []
    iterator = iter(iterable)

    while page_items := [item[0] for item in PageItems(iterator, 2)]:
        pages.append(page_items)

    assert pages == [[1, 2], [3, 4], [5]]

我将无法在 PageItems 上使用 for 循环，因为它最后总是会吐出空白页面，但是使用 while 我可以检查空页，而不必求助于丑陋的 @987654324 @ 堵塞。这也允许我调用list(PageItems(iterator, 2))，如果我只需要没有修改的项目，或者[item[0] for item in PageItems(iterator, 2)]，例如返回的项目是元组，我只想要第一个元素。

【讨论】：