代码之家 › 专栏 › 技术社区 › root-11

将生成器包装为有一个“下一步”调用,而不是两个步骤(uuIter_uu+uuuuu Next_uuuuuu)

generator python-3.x python

root-11 · 技术社区 · 6 年前

我正在从生成器接收用于后台处理的未知数量的记录。如果有更重要的工作,我必须停止发布流程。

这个 main 流程最好描述为:

def main():
    generator_source = generator_for_test_data()  # 1. contact server to get data.
    uw = UploadWrapper(generator_source)  # 2. wrap the data.
    while not interrupt():  # 3. check for interrupts.
        row = next(uw)
        if row is None:
            return
        print(long_running_job(row))  # 4. do the work.

有办法去吗 __next__ 不用插上插头 __iter__ ? 有两个步骤:(1)生成一个迭代器,然后(2)迭代它,这看起来很笨拙。

在许多情况下,我希望将函数提交给函数管理器(MapReduce样式),但在这种情况下,我需要一个具有一些设置的实例化类。因此,仅当单个函数是 _下一个__

class UploadWrapper(object):
    def __init__(self, generator):
        self.generator = generator
        self._iterator = None

    def __iter__(self):
        for page in self.generator:
            yield from page.data

    def __next__(self):
        if self._iterator is None:                # ugly bit.
            self._iterator = self.__iter__()      # 
        try:
            return next(self._iterator)
        except StopIteration:
            return None

问:有更简单的方法吗?

为完整性添加的工作样品:

import time
import random

class Page(object):
    def __init__(self, data):
        self.data = data


def generator_for_test_data():
    for t in range(10):
        page = Page(data=[(t, i) for i in range(100, 110)])
        yield page

def long_running_job(row):
    time.sleep(random.randint(1,10)/100)
    assert len(row) == 2
    assert row[0] in range(10)
    assert row[1] in range(100, 110)
    return row

def interrupt():  # interrupt check
    if random.randint(1,50) == 1:
        print("INTERRUPT SIGNAL!")
        return True
    return False

class UploadWrapper(object):
    def __init__(self, generator):
        self.generator = generator
        self._iterator = None

    def __iter__(self):
        for ft in self.generator:
            yield from ft.data

    def __next__(self):
        if self._iterator is None:
            self._iterator = self.__iter__()
        try:
            return next(self._iterator)
        except StopIteration:
            return None

def main():
    gen = generator_for_test_data()
    uw = UploadWrapper(gen)
    while not interrupt():  # check for job interrupt.
        row = next(uw)
        if row is None:
            return
        print(long_running_job(row))

if __name__ == "__main__":
    main()

1 回复 | 直到 6 年前

Aleksi Torhamo 6 年前

你的 UploadWrapper 似乎过于复杂,有不止一个简单的解决方案。

我的第一个想法是完全放弃课堂,而是使用一个函数:

def uploadwrapper(page_gen):
    for page in page_gen:
        yield from page.data

只需更换 uw = UploadWrapper(gen) 具有 uw = uploadwrapper(gen) 那就行了。

如果你坚持上课,你可以把 __next__() 并更换 uw=上传包装(gen) 具有 uw = iter(UploadWrapper(gen)) ,它会起作用的。

无论哪种情况,您都必须 StopIteration 在呼叫者中。 _下一步 是 想象上的 提高 停止迭代 完成后,不返回 None 就像你的一样。否则,它将无法处理期望一个性能良好的迭代器的事情,例如。 for 循环。

我想你可能对所有的事情应该如何结合有一些误解,所以我会尽我所能解释它应该如何工作,据我所知:

重点 __iter__() 如果你有一个列表,你可以通过调用 iter() . 当你有一个 对于 循环,你首先得到一个迭代器 迭代() 然后打电话 next() 在每个循环迭代中。如果有两个嵌套循环使用相同的列表,那么迭代器及其位置仍然是分开的,因此不会发生冲突。 _ iter_uuuu()号 应该为它所在的容器返回一个迭代器,或者如果它是在迭代器上调用的,则应该只返回 self . 从这个意义上说,这是不对的 上载包装器 不返回 自己 在里面 _ iter_uuuu()号 因为它包装了一个生成器,所以不能真正给出独立的迭代器。至于为什么漏掉 _下一步 有效,因为当你定义一个生成器时 yield 在函数中),生成器具有 _ iter_uuuu()号 (回来了 自己 以及 _下一步 这就是你所期望的。在原始代码中,您没有真正使用 _ iter_uuuu()号 完全是为了它应该被使用的目的:即使你把它重命名为其他的代码也能工作!这是因为你从不打电话 迭代() 在实例上,直接调用 下() .

如果你想在课堂上“适当地”做,我想这样的事情可能就足够了:

class UploadWrapper(object):
    def __init__(self, generator):
        self.generator = generator
        self.subgen = iter(next(generator).data)

    def __iter__(self):
        return self

    def __next__(self):
        while True:
            try:
                return next(self.subgen)
            except StopIteration:
                self.subgen = iter(next(self.generator).data)