python - 使用带有高级运算符的 python 的功能管道

Question

我正在关注 https://youtu.be/R1em4C0oXo8 中的 PyData 演讲，演示者拥有一个用于管道调用的库 yamal。这个库不是开源的。因此，在我用 python 学习 FP 的过程中，我尝试复制该库的基础知识。

简而言之，您在 python 中构建了一系列纯函数(f1、f2、f3, etc) ，并按如下方式创建它们的列表:

pipeline = [f1, f2, f3, f4]

然后，您可以应用函数run_pipeline，结果将是这样的组合:

f4(f3(f2(f1)))

对函数的要求是只有一个返回值，除f1外，都只有一个输入。

这部分很容易实现，我使用组合函数完成了它。

def run_pipeline(pipeline):
    get_data, *rest_of_steps = steps
    def compose(x):
       for f in rest_of_steps:
          y = f(x)
          x = y
       return x

    data = get_data()
    return compose(data)

脱口秀更高级地使用了这个抽象，他定义了“运算符”fork 和reducer。此“运算符(operator)”允许按如下方式运行管道:

pipeline1 = [ f1, fork(f2, f3), f4 ]

相当于:[ f4(f2(f1)), f4(f3(f1)) ]

和

pipeline2 = [ f1, fork(f2, f3), f4, reducer(f5) ]

相当于 f5([f4(f3(f1)), f4(f2(f1))])。

我尝试使用函数式编程 来解决这个问题，但我就是做不到。我不知道 fork 和 reducer 是否是 decorators (如果是，我该如何传递以下函数列表？)不要知道我是否应该使用对象将此列表转换为图形？协程？ (也许这一切都是废话)我简直一头雾水。

有人可以帮助我了解如何使用 python 和函数式编程 构建它吗？

注意:在视频中他谈到了观察者或执行者。对于这个练习，我不关心它们。

Answer 1

虽然这个库旨在促进 Python 中的 FP，但尚不清楚库本身是否应该使用大量 FP编写。

这是一种使用类(基于list类型)告诉pipe函数是否需要fork或reduce，是否正在处理的实现方式使用单个数据项或项目列表。

这使得 FP 样式技术的使用受到一些限制，例如对 apply_func 的递归调用(允许在管道中进行多个分支)。

class Forked(list):
    """ Contains a list of data after forking """

class Fork(list):
    """ Contains a list of functions for forking """

class Reducer(object):
    """ Contains a function for reducing forked data """
    def __init__(self, func):
        self.func = func

def fork(*funcs):
    return Fork(funcs)

def reducer(func):
    """ Return a reducer form based on a function that accepts a
        Forked list as its first argument """
    return Reducer(func)   

def apply_func(data, func):
    """ Apply a function to data which may be forked """
    if isinstance(data, Forked):
        return Forked(apply_func(datum, func) for datum in data)
    else:
        return func(data)

def apply_form(data, form):
    """ Apply a pipeline form (which may be a function, fork, or reducer) 
        to the data """
    if callable(form):
        return apply_func(data, form)
    elif isinstance(form, Fork):
        return Forked(apply_func(data, func) for func in form)
    elif isinstance(form, Reducer):
        return form.func(data)

def pipe(data, *forms):
    """ Apply a pipeline of function forms to data """
    return reduce(apply_form, forms, data)

使用示例:

def double(x): return x * 2
def inc(x): return x + 1
def dec(x): return x - 1
def mult(L): return L[0] * L[1]

print pipe(10, inc, double)    # 21
print pipe(10, fork(dec, inc), double)  # [18, 22]
print pipe(10, fork(dec, inc), double, reducer(mult))   # 396

编辑:这也可以通过使 fork 成为一个返回函数的函数和 reducer 一个创建模仿函数的对象的类来进一步简化。然后不再需要单独的 Fork 和 Reducer 类。

class Forked(list):
    """ Contains a list of data after forking """

def fork(*funcs):
    """ Return a function that will take data and output a forked
    list of results of putting the data through several functions """
    def inner(data):
        return Forked(apply_form(data, func) for func in funcs)

    return inner

class reducer(object):
    def __init__(self, func):
        self.func = func

    def __call__(self, data):
        return self.func(data)

def apply_form(data, form):
    """ Apply a function or reducer to data which may be forked """
    if isinstance(data, Forked) and not isinstance(form, reducer):
        return Forked(apply_form(datum, form) for datum in data)
    else:
        return form(data)

def pipe(data, *forms):
    """ Apply a pipeline of function forms to data """
    return reduce(apply_form, forms, data)