在Python中表达式地组合生成器

Question

我非常喜欢Python生成器。特别是，我发现它们是连接Rest端点的理想工具 - 我的客户端代码只需迭代与端点相连的生成器即可。然而，我发现Python的生成器在一个领域不如我所希望那样表达。通常，我需要过滤从端点获取的数据。在我的当前代码中，我将谓词函数传递给生成器，它将谓词应用于正在处理的数据，仅在谓词为True时产生数据。

我希望向生成器组合的方向发展 - 如 data_filter(datasource())。以下是一些演示代码，显示了我尝试过的内容。很明显，它为什么无法工作，我正在努力找到最富有表现力的方法来解决问题：

# Mock of Rest Endpoint: In actual code, generator is 
# connected to a Rest endpoint which returns dictionary(from JSON).
def mock_datasource ():
    mock_data = ["sanctuary", "movement", "liberty", "seminar",
                 "formula","short-circuit", "generate", "comedy"]
    for d in mock_data:
        yield d

# Mock of a filter: simplification, in reality I am filtering on some
# aspect of the data, like data['type'] == "external" 
def data_filter (d):
    if len(d) < 8:
        yield d

# First Try:
# for w in data_filter(mock_datasource()):
#     print(w)
# >> TypeError: object of type 'generator' has no len()

# Second Try 
# for w in (data_filter(d) for d in mock_datasource()):
#     print(w)
# I don't get words out, 
# rather <generator object data_filter at 0x101106a40>

# Using a predicate to filter works, but is not the expressive 
# composition I am after
for w in (d for d in mock_datasource() if len(d) < 8):
    print(w)

- chladni

---

你对内置的 filter() 函数有什么感觉？ - Kevin

好的建议 - 如果我使用谓词函数，我会写成filter(data_predicate, mock_datasource())。然而，我更喜欢能够像f(g(x))这样编写生成组合的方法。 - chladni

在这种情况下，@Kevin需要使用lambda来调用filter，这样你就会得到一个笨重的表达式。当过滤函数已经存在时（例如str.isdigit、None用于测试真值等），filter是很好的选择。 - Jean-François Fabre

@Jean-FrançoisFabre，同意，“filter”是一种“有时”的解决方案。这就是为什么我没有花费太多精力来构建一个完整的答案 :-P - Kevin

在Python 2中，filter在字符串上非常有用，因为它可以避免使用str.join。现在这种乐趣已经消失了 :) - Jean-François Fabre

Answer 1

data_filter应该对d的元素应用len而不是对d本身进行操作，像这样：

def data_filter (d):
    for x in d:
        if len(x) < 8:
            yield x

现在你的代码：

for w in data_filter(mock_datasource()):
    print(w)

返回值

liberty
seminar
formula
comedy

- Jean-François Fabre

---

谢谢，这似乎是最接近我所要求的。话虽如此，我想知道编写生成器是否会带来我没有考虑到的性能成本。 - chladni

没错，你越是链式调用函数/生成器，你的应用程序就会越慢。在Python中调用函数比编译语言更昂贵，部分原因是编译语言有能力内联一些调用。 - Jean-François Fabre

到目前为止，在测试中比较使用谓词过滤和使用组合生成器进行过滤的执行时间（即基于您的答案），我没有看到组合方法有很大的性能惩罚。通常情况下，需要运行更多的测试。"第一个原则是你不要欺骗自己，因为你是最容易被欺骗的人。" 理查德·费曼 - chladni

这是真的。你需要使用相关的数据大小（大小和内容）来测试各种方法。 - Jean-François Fabre

Answer 2

更简洁地说，您可以直接使用生成器表达式来完成此操作：

def length_filter(d, minlen=0, maxlen=8):
    return (x for x in d if minlen <= len(x) < maxlen)

像普通函数一样将过滤器应用于您的生成器：

for element in length_filter(endpoint_data()):
    ...

如果你的谓词非常简单，内置函数 filter 也许可以满足你的需求。

- wim

Answer 3

这是我一直在使用的将生成器组合在一起的函数。

def compose(*funcs):
    """ Compose generators together to make a pipeline.
    e.g.
        pipe = compose(func1, func2, func3)
        result = pipe(range(0, 5))
    """
    return lambda x: reduce(lambda f, g: g(f), list(funcs), x)

funcs 是一个生成器函数列表。因此，您的示例将如下所示：

pipe = compose(mock_datasource, data_filter)
print(list(pipe))

这不是原创

- CpILL

Answer 4

您可以传递一个过滤函数，对每个项目应用该函数：

def mock_datasource(filter_function):
    mock_data = ["sanctuary", "movement", "liberty", "seminar",
             "formula","short-circuit", "generate", "comedy"]

    for d in mock_data:
        yield filter_function(d)

def filter_function(d):
    # filter
    return filtered_data

- meow

---

没错 - 你提出的方法与我正在使用且有效的代码类似。我试图将过滤器放置在数据源的输出端。我想完全将过滤器从生成器的代码中剥离出来。我所能接近的是我给出的最后一个示例中使用谓词。无论如何，感谢你的建议！ - chladni

Answer 5

我会定义一个名为filter(data_filter)的函数，该函数接收一个生成器作为输入，并返回一个由data_filter谓词（常规谓词，不了解生成器接口）过滤值的生成器。

代码如下：

def filter(pred):
    """Filter, for composition with generators that take coll as an argument."""
    def generator(coll):
        for x in coll:
            if pred(x):
                yield x
    return generator

def mock_datasource ():
    mock_data = ["sanctuary", "movement", "liberty", "seminar",
                 "formula","short-circuit", "generate", "comedy"]
    for d in mock_data:
        yield d

def data_filter (d):
    if len(d) < 8:
        return True


gen1 = mock_datasource()
filtering = filter(data_filter)
gen2 = filtering(gen1) # or filter(data_filter)(mock_datasource())

print(list(gen2)) 

如果您想进一步提高，可以使用 compose，我认为这是整个意图：

from functools import reduce

def compose(*fns):
    """Compose functions left to right - allows generators to compose with same
    order as Clojure style transducers in first argument to transduce."""
    return reduce(lambda f,g: lambda *x, **kw: g(f(*x, **kw)), fns)

gen_factory = compose(mock_datasource, 
                      filter(data_filter))
gen = gen_factory()

print(list(gen))

附注：我使用了在这里找到的一些代码，其中Clojure的人表达了由生成器组成的方式，这种方式受到他们使用转换器进行通用组合的启发。 附注2：filter可以以更Pythonic的方式编写：

def filter(pred):
    """Filter, for composition with generators that take coll as an argument."""
    return lambda coll: (x for x in coll if pred(x))

- dqc