在Python中，如何将项目逐一推送到另一个函数使用的迭代器？

Question

假设我有一个现有的函数，它消耗一个可迭代对象，例如：

def printall(iterable):
    for item in iterable:
        print(item)

重点是，该函数已经存在，我不想克隆或修改它。

但我想给它一些重要的处理。伪代码：

push_handle = XXXX()
printall(push_handle)
# printall() is now waiting on next(iter(iterable))

for item0 in some_source():
    statement
    statement
    item1 = some_transformation(item0)
    push_handle.send(item1)
    # at this point printall() iterates once and resumes waiting

我知道我可以将上面的循环重构为生成器本身：

def processing(iterable):
    for item0 in iterable:
        statement
        statement
        item1 = some_transformation(item0)
        yield item1

printall(processing(some_source()))

即便如此，我觉得上面概述的形式的“适配器”会很有用，无需重构。

我使用过Python生成器，但我一直被这样一个事实所困扰：在生成器函数中，

yield

隐式地屈服于原始调用者。我想象一个包含两个生成器方法的类，一个的

yield

表达式接收

send()

调用，另一个

yield

到其他调用者 - 但将控制从一个方法转移到另一个方法是断开连接。

我可以使用线程和队列来做到这一点，但这对于这个目的来说感觉太重了。

同样，我可以使用 eventlet.queue.Queue(0)，但生产者和消费者必须在不同的 eventlet greenthreads 上运行。

这感觉好像应该像编码一样简单

yield

，但我被困住了。我是不是忽略了什么？

Answer 1

在您的示例中，您可以简单地使用这个：

for item0 in some_source():
    statement
    statement
    item1 = some_transformation(item0)
    print(item1)
    push_handle.send(item1)

...但为了更通用的目的，我认为你应该使用观察者模式：

class IDPrinter:

    def __call__(item):
        print(id(item))


class XXXX:

    def __init__(self):
        self._observables = []

    def add_observable(self, obs: Callable):
        self._observable.append(obs)

    def send(item):
        for observable in self._observables:
            observable(item)
        # Do something


push_handle = XXXX()
push_handle.add(print, IDPrinter())

for item0 in some_source():
    statement
    statement
    item1 = some_transformation(item0)
    push_handle.send(item1)

在这种情况下，您可以将

Observable

替换为任何

Callable

。只要传递的项目可以由可调用函数处理，就可以了。

如果你真的不想重构你的

XXXX

类，你可以使用事件循环。这不是一种非常有效的方法，因为您必须不断检查迭代器中是否有某些内容，这会浪费计算资源。如果我是你，我会选择观察者模式。希望有帮助。