表示可多次迭代的 Iterable 的 Python 方式是什么

Question

我想得到你关于用类型提示在 python 中表达以下函数的最 Pythonic 方式的建议：

我想公开一个函数作为库的一部分，该库接受输入参数并返回输出。输入参数的约定应该是：

• 我的函数可以迭代它
• 如果我的函数维护对输入的引用（例如，通过返回保留该引用的对象），那就可以了
• 多次迭代输入是可以的

一个示例可能是一个函数，它接受一系列 URL，然后向这些 URL 发出请求，可能带有一些重试逻辑，因此我必须多次迭代原始序列。但我的问题比这个示例更通用。

乍一看，合适的签名是：

from typing import Iterable

def do_sth(input: Iterable[str]) -> SomeResult:
  ...

然而这违反了第三个要求，因为在Python中不能保证你可以多次迭代一个Iterable，例如因为迭代器和生成器本身就是可迭代的。

另一种尝试可能是：

from typing import Sequence

def do_sth(input: Sequence[str]) -> SomeResult:
  ...

但是

Sequence

合约超出了我的功能所需，因为它包括索引访问以及长度知识。

我想到的一个解决方案是使用

Iterable

签名，然后在内部复制输入。但是，如果源序列很大，这似乎会引入潜在的内存问题。

有没有解决这个问题的方法，即Python是否知道每次都会返回一个新迭代器的

Iterable

的概念？

Answer 1

我能想到有两种自然的方式来表示这一点。

第一种方法是使用

Iterable[str]

，并在文档中提到，不应使用

Iterator

和

Generator

对象，因为您可能会多次调用

__iter__

。

Iterable

的全部意义在于你可以在其上获得迭代器，可以说，首先让

Iterator

支持

Iterable

是一个错误。它并不完美，但很简单，通常比技术上更正确但非常复杂的注释更“Pythonic”。

您可以添加一些运行时检查，如果用户传递了错误的内容，则会提醒用户存在问题：

iter1 = iter(input)
for item in iter1:
    do_something(item)
iter2 = iter(input)
if iter2 is iter1:
    raise ValueError("Must pass an iterable that can be iterated multiple times. Got {input}.")

或者检查你是否有 Iterator，并用内存惩罚来处理它：

if isinstance(input, Iterator):
    input = list(input)  # or itertools.tee or whatever
    warn("This may eat up a lot of memory")

另一种选择是使用

io.TextIOBase

。这可以通过查找开头来迭代多次。这取决于您的用例，并且可能不太适合。如果从概念上讲，输入是字符序列上的某种分块视图，那么 io 流就很适合，即使迭代器在技术上不返回文本行。如果它在概念上是一个不连续的字符串序列，那么流就不适合。

Answer 2

您可以使用不接受输入并返回可迭代的函数。在输入提示方面，您可以使用

Callable

。

从文档中，如果您不熟悉

Callable

：

需要特定签名的回调函数的框架可能会使用

Callable[[Arg1Type, Arg2Type], ReturnType]

进行类型提示。

解决方案：

from typing import Callable, Iterable

def do_sth(get_input: Callable[[], Iterable[str]]) -> SomeResult:
    # ...
    pass

def main():
    do_sth(lambda : (str(i) for i in range(10)))

我的函数可以迭代它

def do_sth(get_input: Callable[[], Iterable[str]]) -> SomeResult:
    for item in get_input():
        pass

如果我的函数维护对输入的引用（例如，通过返回保留该引用的对象），那就可以了

不明白为什么不。

def do_sth(get_input: Callable[[], Iterable[str]]) -> SomeResult:
    return dict(reference=get_input)

多次迭代输入是可以的

def do_sth(get_input: Callable[[], Iterable[str]]) -> SomeResult:
    for i in range(10**82):
        for item in get_input():
            pass

Answer 3

我开发了一个包

python-repeatable-iterable

，它提供了RepeatableIterable类型和获取此类类型的函数。

https://pypi.org/project/python-repeatable-iterable/

https://github.com/LLyaudet/python-repeatable-iterable

例如，对于 Django，您可以像这样使用它：

from django.db.models import QuerySet

tasks = get_repeatable_iterable(tasks, (QuerySet,))

目前只有列表和元组被“原生”视为可重复迭代。 如果您在 Python 标准库中看到其他有用的案例，请提交 PR。