python 中的方法链延迟执行

method chain lazy execution in python

我在 TestClass 对象的 python 程序中有一系列方法,如下所示:-

TestClass().method1(args).method2(args).method3(args)

我想构建一个实用程序来扫描上面的链,然后它可以构建一个执行序列。 (类似于数据处理框架构建其执行图或 dag 的方式)。此外,在我最后调用 run() 之前,上面的序列不会 运行。

所以,在我的例子中,只有当 run() 方法被添加到链的末尾时,它才会 运行。

TestClass().method1(args).method2(args).method3(args).run()

如何实现?

我从这个 answer 中得到启发,以便详细说明我的。

因此,我将使用 inspectre 模块以及 decorator 来检查是否调用了 run 方法:

import re
import inspect

def check_run_is_called(func):
    def wrapper(*args, **kwargs):
        # Here the interesting part
        # We're searching if the string `.run()` exists in the code context
        run_exist = re.search(r'\.run\(.*?\)', inspect.stack()[-1].code_context[0])
        if not run_exist:
            raise Exception('Run must be called !')
        f = func(*args, **kwargs)
        return f
    return wrapper


class TestClass():
    @check_run_is_called
    def method1(self, arg):
        print(arg)
        return self

    @check_run_is_called
    def method2(self, arg):
        print(arg)
        return self

    @check_run_is_called
    def method3(self, arg):
        print(arg)
        return self

    def run(self):
        print('Run')


# test
if __name__ == '__main__':
    app = TestClass()
    app.method1('method1').method2('method2').method3('method3').run()

输出:

method1
method2
method3
Run

否则,如果我们调用不包含run方法的方法链:

app.method1('method1').method2('method2').method3('method3')

我们会得到一个 Exception:

Exception: Run must be called !

除此之外,您还可以像这样创建链的组合:

app.method1('method1').run()
# method1
# Run
app.method1('method1').method2('method2').run()
# method1
# method2
# Run
app.method2('method2').method3('method3').method1('method1').run()
# method2
# method3
# method1
# Run

你想要的是流畅的界面。

我会给你一个简单的例子,肯定可以用某种装饰器做得更好

class A:
    def __init__(self):
        self.call_chain = []

    def m1(self, arg):
        def _actual_logic():
            print(arg)
        self.call_chain.append(actual_logic)
        return self

    def run(self):
        for f in self.call_chain:
            f()

a = A().m1('snth')
a.run()