代码之家 › 专栏 › 技术社区 › nanounanue

使用python和高级操作符的函数管道

functional-programming lambda function python

nanounanue · 技术社区 · 6 年前

我正在关注PyData的谈话 https://youtu.be/R1em4C0oXo8 ,演示者创建用于管道调用的库 yamal 。此库不是开源的。所以,在我学习FP的方式中 python ,我试图复制 基础知识 那个图书馆的。

简而言之,您构建了一系列 纯净的 中的函数 python ( f1 ,则, f2 ,则, f3 ,并创建如下列表:

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 “操作员”允许按照以下方式运行管道:

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))]) 。

我尝试使用 函数式编程 ,但我不能。我不知道 叉 和 减速器 是 decorators (如果是,我如何传递以下函数列表?)不知道是否应该使用对象将此列表转换为图形?协同程序?(也许这一切都是胡说八道)我简直完全糊涂了。

有人能帮我用 python 和 函数式编程 ?

笔记 :在视频中,他谈到了观察员或遗嘱执行人。对于这个练习,我不在乎他们。

1 回复 | 直到 6 年前

Stuart 6 年前

虽然此库旨在促进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)