0

我有一个想要执行的函数列表，比如在管道中。

pipeline = [f1, f2, f3, f4]

（基本上我编写列表中的函数，请参阅：https ://mathieularose.com/function-composition-in-python/ ）：

def run(pipeline):
     return functools.reduce(lambda f, g: lambda x: f(g(x)), pipeline, lambda x: x)

我想观察函数何时被调用、何时完成、是否失败等，并将其记录到文件或数据库或 MongoDB 等。每个函数都返回一些 python 对象，所以我想使用返回值并记录其属性，例如

如果f1返回一个list我想登录f1 was completed at 23:00 on 04/22/2018. It returned a list of length 5等等

我的问题不是关于执行函数，而是关于观察函数的行为。我希望这些功能与我如何编码管道无关。

我想知道如何在这里实现观察者模式。我知道“观察者模式”听起来太“面向对象”了，所以我的第一个想法是使用装饰器，但在寻找这方面的指南时我发现了RxPy.

因此，我正在寻找有关如何解决此问题的指南。

4

2 回答 2

1

就像是：

import time

def run(pipeline):
    def composed(x):
        for f in pipeline:
            y = f(x)
            print('{} completed on {}. it returned {}'.format(f, time.time(), y))
            x = y
        return x
    return composed

观察者模式不太适合函数式风格。

编辑：基于回调：

def run(pipeline, callback):
    def composed(x):
        for f in pipeline:
            y = f(x)
            callback(f, time.time(), y)
            x = y
        return x
    return composed
于 2018-04-23T01:30:09.800 回答
1

如果您喜欢装饰器，可以使用以下方法：

def log_finish_time(f):
    def g(*args):
        ret = f(*args)
        print('{} completed on {}'.format(f, time.time()))
        return ret
    return g

def log_return_value(f):
    def g(*args):
        ret = f(*args)
        print('{} returned {}'.format(f, ret))
        return ret
    return g

@log_finish_time
@log_return_value
def f1(x):
    return x+1

@log_finish_time
@log_return_value
def f2(x):
    return x*x

无需更改您的合成代码。

如果你想为你的装饰器添加参数，你必须在中间添加另一个函数（基本上它是一个返回装饰器的函数）：

def log_finish_time(log_prefix):
    def h(f):
        def g(*args):
            ret = f(*args)
            print('{}: {} completed on {}'.format(log_prefix, f, time.time()))
            return ret
        return g
    return h

@log_finish_time('A')
def f1(x):
    return x+1
于 2018-04-23T01:35:45.840 回答