链接嵌套产量

Question

我想构建一个数据管道，对数据系列的行执行一系列操作。

大多数函数将在一行输入，一行输出的基础上工作，但其中一些操作将“扩展”系列 - 我的意思是一行将进入函数并且可能生成多行作为该功能的结果。

我想设置一系列足够强大的函数来自己处理这种行为，而不必编写一堆疏忽代码。

使用yield作为一个机会 - 如果每个函数都消耗了前一个函数的产量，并且自己充当了一个生成器，那么我可以任意地将一堆这些格式良好的函数链接在一起 - 这从一个优雅的角度来看是很好的。视图。

这是我的设置代码，func_x充当简单的1-1功能，func_y进行扩展。

from collections import OrderedDict
data_source = [ OrderedDict({"id" : "1", "name" : "Tom", "sync" : "a"}),
            OrderedDict({"id" : "2", "name" : "Steve", "sync" : "a"}),
            OrderedDict({"id" : "3", "name" : "Ulrich", "sync" : "b"}),
            OrderedDict({"id" : "4", "name" : "Victor", "sync" : "b"}),
            OrderedDict({"id" : "5", "name" : "Wolfgang", "sync" : "c"}),
            OrderedDict({"id" : "6", "name" : "Xavier", "sync" : "c"}),
            OrderedDict({"id" : "7", "name" : "Yves", "sync" : "c"}),
            OrderedDict({"id" : "8", "name" : "Zaphod", "sync" : "d"})]
def row_getter(source):
    for content in source:
        yield content.copy()

def func_x(row):
    try:
        q=next(row)
        if q['name']=="Tom":
            q['name']="Richard"
        yield q.copy()
    except StopIteration:
        print ("Stop x")


def func_y(row):
    try:
        q=next(row)
        for thingy in range(0,2):
            q['thingy']=thingy
            yield q.copy()
    except StopIteration:
        print ("Stop y")

rg = row_getter(data_source)
iter_func = func_y(func_x(rg))

现在，我可以通过迭代iter_func对象来获取第一组数据：

print (next(iter_func))
>> OrderedDict([('id', '1'), ('name', 'Richard'), ('sync', 'a'), ('thingy', 0)])

然后再次：

print (next(iter_func))
>> OrderedDict([('id', '1'), ('name', 'Richard'), ('sync', 'a'), ('thingy', 1)])

而且，虽然这一次，而不是看到史蒂夫的记录（即流程中的下一个记录，现在第一个记录上func_y的扩展已经完成），我得到一个StopIteration错误。

print (next(iter_func))
>> StopIteration                             Traceback (most recent call last)
<ipython-input-15-0fd1ed48c61b> in <module>()
----> 1 print (next(iter_func))

StopIteration: 

所以我不明白这是从哪里来的，因为我试图在func_x和func_y陷阱。

Answer 1

你的func_x函数只生成一个项目，所以它会在消耗完之后完成。试试这样的事情：

def func_x(row):
    try:
        for q in row:
            if q['name']=="Tom":
                q['name']="Richard"
            yield q
    except StopIteration:
        print ("Stop x")

顺便说一下，请注意每个屈服都不会制作对象的副本。在许多情况下这可能没什么问题，但请注意，在func_y中，您将产生两次相同的对象，将'thingy'设置为不同的值。这意味着，例如，如果您这样做（在您发布的代码之后）：

d1 = next(iter_func)
d2 = next(iter_func)

d1和d2将是同一个对象，特别是他们都将'thingy'设置为1。

Answer 2

内置工具（特别是map和itertools.chain）可以为您做到这一点。

from collections import OrderedDict
from itertools import chain


data_source = [ OrderedDict({"id" : "1", "name" : "Tom", "sync" : "a"}),
            OrderedDict({"id" : "2", "name" : "Steve", "sync" : "a"}),
            OrderedDict({"id" : "3", "name" : "Ulrich", "sync" : "b"}),
            OrderedDict({"id" : "4", "name" : "Victor", "sync" : "b"}),
            OrderedDict({"id" : "5", "name" : "Wolfgang", "sync" : "c"}),
            OrderedDict({"id" : "6", "name" : "Xavier", "sync" : "c"}),
            OrderedDict({"id" : "7", "name" : "Yves", "sync" : "c"}),
            OrderedDict({"id" : "8", "name" : "Zaphod", "sync" : "d"})]


def rename(d):
    if d['name'] == "Tom":
        d['name'] = "Richard"
    return d


def add_thingy(d):
    for y in range(2):
        yield {'thingy': y, **d}

for x in chain.from_iterable(add_thingy(d) 
                             for d in map(rename,
                                          data_source)):
    print(x)

map不是必需的;我们可以将rename应用到每个dict，然后再将它们传递给生成器表达式中的add_thingy。

for x in chain.from_iterable(add_thingy(rename(d)) for d in data_source):
    print(x)

或者走另一条路并使用map两次：

for x in chain.from_iterable(map(add_thingy, map(rename, data_source))):
    print(x)