Pythonic 方式链接 python 生成器函数以形成管道答案

【问题标题】：Pythonic way to chain python generator function to form a pipelinePythonic 方式链接 python 生成器函数以形成管道
【发布时间】：2016-12-09 21:21:28
【问题描述】：

我正在使用 python 进行管道代码重构。

假设我们有一系列 generator 函数，我们希望将它们链接起来以形成一个数据处理管道。

例子：

#!/usr/bin/python
import itertools

def foo1(g):
    for i in g:
        yield i + 1

def foo2(g):
    for i in g:
        yield 10 + i

def foo3(g):
    for i in g:
        yield 'foo3:' + str(i)

res = foo3(foo2(foo1(range(0, 5))))

for i in res:
    print i

输出：

foo3:11
foo3:12
foo3:13
foo3:14
foo3:15

我不认为foo3(foo2(foo1(range(0, 5)))) 是实现我的管道目标的pythonic 方式。尤其是当管道中的阶段数很大时。

我希望我可以像 jquery 中的链一样重写它。类似于：

range(0, 5).foo1().foo2().foo3()

或许

l = [range(0, 5), foo1, foo2, foo3]
res = runner.run(l)

但我是生成器主题的新手，找不到实现此目的的方法。

欢迎任何帮助。

【问题讨论】：

也许是itertools.accumulate 或functools.reduce？（两者都是标准库的一部分）
这听起来像是一些 XY 问题。如果您正在对数组/列表进行数字运算，请考虑使用 numpy/pandas。
maxymoo 的答案可能是最好的，但您也可以以不正当的方式滥用运算符重载（这在 Python 中是不受欢迎的），请参阅此以获得灵感：stackoverflow.com/questions/33658355/…

标签： python generator

【解决方案1】：

我有时喜欢在这种情况下使用左折叠（在 Python 中称为 reduce）：

from functools import reduce
def pipeline(*steps):
    return reduce(lambda x, y: y(x), list(steps))

res = pipeline(range(0, 5), foo1, foo2, foo3)

甚至更好：

def compose(*funcs):
    return lambda x: reduce(lambda f, g: g(f), list(funcs), x)

p = compose(foo1, foo2, foo3)
res = p(range(0, 5))

【讨论】：

除了比链式foo(foo1(...) 效率低之外，即使有很多函数，这也不是我所说的更具可读性。
这只是个人喜好问题，您可以像 @john1024 那样为它定义一个别名，但对我来说，我只是将模式识别为管道，并专注于函数列表作为有意义的部分，主要是它们是从左到右而不是从右到左流水线，就像问题中的函数组合一样
@MSeifert - 这效率如何降低？我觉得它同样有效。
@MSeifert：现在想象一下，当你有 10-20 个不同的生成器，它们的名称实际上是描述性的。
是的，如果对大型数据集进行操作，效率不会受到影响。我不认为答案是错误的或不推荐的。我只是想指出，在我看来从长远来看，像 foo1(foo2(...)) 这样的东西更具可读性。

【解决方案2】：

按照您的 runner.run 方法，让我们定义这个实用函数：

def recur(ops):
    return ops[0](recur(ops[1:])) if len(ops)>1 else ops[0]

举个例子：

>>> ops = foo3, foo2, foo1, range(0, 5)
>>> list( recur(ops) )
['foo3:11', 'foo3:12', 'foo3:13', 'foo3:14', 'foo3:15']

替代方案：倒序

def backw(ops):
    return ops[-1](backw(ops[:-1])) if len(ops)>1 else ops[0]

例如：

>>> list( backw([range(0, 5), foo1, foo2, foo3]) )
['foo3:11', 'foo3:12', 'foo3:13', 'foo3:14', 'foo3:15']

【讨论】：

如果你有超过 1000 个函数，它就不起作用了！
我认为以相反的顺序排列功能是混乱的。 return ops[-1](recur(ops[:-1])) if len(ops)>1 else ops[0] 呢？
@rbierman 非常好。是的，这也有效。使用反向排序代码更新答案。

【解决方案3】：

如果您的示例中的功能是一次性（或一次性）功能，这是另一个答案。一些不错的变量命名和生成器表达式的使用对小型操作很有帮助。

>>> g = range(0, 5)
>>> foo1 = (x+1 for x in g)
>>> foo2 = (x+10 for x in foo1)
>>> foo3 = ('foo3:' + str(x) for x in foo2)
>>> for x in foo3:
...     print x
...
foo3:11
foo3:12
foo3:13
foo3:14
foo3:15

【讨论】：

【解决方案4】：

您可以使用PyMonad 编写柯里化生成器函数：

def main():
    odds = list * \
         non_divisibles(2) * \
         lengths * \
         Just(["1", "22", "333", "4444", "55555"])
    print(odds.getValue())    #prints [1, 3, 5]


@curry
def lengths(words: Iterable[Sized]) -> Iterable[int]:
    return map(len, words)


@curry
def non_divisibles(div: int, numbers: Iterable[int]) -> Iterable[int]:
    return (n for n in numbers if n % div)

另一种选择是从 Monad 开始并使用 fmap 调用组成生成器 - Java 8 Stream 用户熟悉这种语法：

def main():
    odds = Just(["1", "22", "333", "4444", "55555"]) \
        .fmap(lengths) \
        .fmap(non_divisibles(2)) \
        .fmap(list) \
        .getValue()
    print(odds)   #prints [1, 3, 5]


def lengths(words: Iterable[Sized]) -> Iterable[int]:
    return map(len, words)


@curry
def non_divisibles(div: int, numbers: Iterable[int]) -> Iterable[int]:
    return (n for n in numbers if n % div)

请注意，在这种情况下，函数不需要用@curry 修饰。在终端 getValue() 调用之前，不会评估整个转换链。

【讨论】：

【解决方案5】：

我不认为 foo3(foo2(foo1(range(0, 5)))) 是实现我的管道目标的 Python 方式。尤其是当管道中的阶段数很大时。

有一种相当简单但在我看来很清楚的链接生成器的方法：将每个生成器的结果分配给一个变量，其中每个都可以有一个描述性名称。

range_iter = range(0, 5)
foo1_iter = foo1(range_iter)
foo2_iter = foo2(foo1_iter)
foo3_iter = foo3(foo2_iter)

for i in foo3_iter:
  print(i)

我更喜欢这个而不是使用更高阶函数的东西，例如reduce 或类似名称：

在我的实际案例中，通常每个 foo* 生成器函数都需要自己的其他参数，如果使用 reduce，这会很棘手。
在我的真实案例中，管道中的步骤在运行时不是动态的：（对我来说）有一个似乎更适合动态案例的模式似乎有点奇怪/出乎意料。
这与常规函数的通常编写方式有点不一致，其中每个函数都被显式调用，并且每个函数的结果都传递给下一个函数的调用。是的，我猜有点重复，但我很高兴“调用函数”被重复，因为（对我来说）它真的很清楚。
无需导入：它使用核心语言功能。

【讨论】：

【解决方案6】：

对于未来的读者：另一个非常 Pythonic 的解决方案（恕我直言）：

steps = [
    foo1, 
    foo2, 
    foo3
    ]

res = range(0, 5)
for step in steps:
    res = step(res)

for i in res:
    print(i)

foo3:11
foo3:12
foo3:13
foo3:14
foo3:15

这与 maxymoo 的回答中的 functools.reduce 基本上是一样的。生成器的懒惰使得这个简单的公式无需 functools。

【讨论】：