1

我尝试使用 设计工作流composite pattern，示例代码如下所示：

class CommandInterface(object):
    def __init__(self, name, template=None, tool=None, param : dict={},input=None, output=None ):
        self.name = name
        self.input = input
        self.output = output
        self.param = param
        self.template = template
    def before_invoke(self):
        pass   # check before invoke
    def invoke(self):
        pass
    def after_invoke(self):
        pass  # check after invoke
class ShellCommand(CommandInterface):    
    def render(self):
        cmd = self.template.format(input=self.input,
                                    output=self.output,
                                    param=self.param,
                                    )
        return cmd
    def before_invoke(self):
        pass
    def invoke(self):
        os.system(self.render())

class Workflow(CommandInterface):
    def __init__(self, name):
       self.name = name
       self._input = []
       self._output = []
       self._commands = []
       self._tool = []
    def add(self, command : CommandInterface):
        self._commands.append(command)
        return self

    def invoke(self):
        for command in self._commands:
            command.invoke()

并且可以这样使用：

workflow = Workflow()
raw_files = ["file1", "file2", "file3"]
for i in raw_files:
    head_command = ShellCommand("head {input} {output}", 
                                input = i, 
                                output = i+"_head")
    workflow.add(head_command)
merge_command = ShellCommand("cat {input} > {output}", 
                             input=" ".join([i+"_head" for i in rawfiles]), 
                             output="merged")
workflow.add(merge_command)

workflow.invoke()

但是，有时一个命令的输入可能是另一个命令的结果，例如：

class PythonCommand(CommandInterface):    
    def invoke(self):
        self._result = self.template(input=self.input, output=self.output, param=self.param)

def a_command(input, output, param):
    take_long_time(input, output, param)
    return {"a_result": "some result will be used in b_command"}

def b_command(input, output, param):
    def need_the_result_of_a_command(input, output, param):
        return param["a_result"]
    need_the_result_of_a_command(input, output, param)
    return "success"

a = PythonCommand(a_command, input_, output_, param_)
a.invoke()
b = PythonCommand(b_command, input_, output_, param= a._result)
b.invoke()

我不能使用工作流来合成这 2 个命令，因为b使用a' 结果作为参数，这仅在调用 a 后可见。但是，在某些情况下，仍然需要像之前的代码这样的工作流管理器。我想我需要这样的东西：

workflow = Workflow()
a = PythonCommand(a_command, input_, output_, param_)
workflow.add(a)
b = PythonCommand(b_command, input_, output_, param= {"a_result": a.lazy()._result})
workflow.add(b)
workflow.invoke()

有没有人对如何设计具有.lazy()这样实现的工作流有想法？

4

0 回答 0