Pythonic方式表示可迭代对象可多次迭代

Question

我希望能得到你的建议，关于在Python中使用类型提示表达以下函数的最佳方式：

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

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

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

乍一看，一个适当的签名将是：

from typing import Iterable

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

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

另一种尝试可能是：

from typing import Sequence

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

但是Sequence协议比我所需的函数更多，因为它包括索引访问和长度的知识。

我想到的一个解决方案是使用Iterable签名，然后在内部复制输入。但是如果源序列很大，这似乎会引入潜在的内存问题。

是否有解决方法，即Python是否知道每次都会返回新迭代器的Iterable的概念？

- Carsten

---

我不会过于深思熟虑这个问题。从实际目的来看，点2和点3意味着你需要一个列表，或者将你得到的任何内容转换为列表。只需将输入类型设为List[str]，让调用者担心如何从可能不可迭代的值中创建一个列表即可。 - chepner

考虑 itertools.cycle：foo2 = cycle(foo) 内部缓存了foo的元素，以便可以无限次地重复它们，但是foo和foo2不再是独立的；在一个上调用next会消耗另一个上的项目。另一个例子是itertools.tee；文档明确表示，在f1，f2 = tee(foo)之后不能可靠地使用foo；您必须使用f1和f2。 - chepner

然而，这违反了第三个要求，因为在Python中，不能保证您可以多次迭代Iterable，例如，因为迭代器和生成器本身就是可迭代的。我认为这强制输入是不可变的->元组将是正确的选择。 - Pablo Henkowski

@chepner 谢谢，我想在这种情况下我需要一个Tuple，因为我希望它是不可变的。我对这种方法的困扰是，这将在接口中使用实现类而不是抽象类。 - Carsten

“Collection”可能是你能得到的最接近的词，比“Sequence”更好，因为它不需要可索引。 - levsa

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}.")

或者检查您是否获得了迭代器，并通过内存惩罚进行处理：

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

另一种选择是使用 io.TextIOBase。通过将指针移动到开头，可以多次迭代。这取决于你的使用情况，并不一定适合。如果概念上输入是对一系列字符的分块视图，则 io 流是一个很好的选择，即使迭代器在技术上并不返回文本行。如果它的概念是一系列不连续的字符串，则流不适用。

- Lucas Wiman

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

- oglehb