Thread
#prefect-community
    j

    Jan Feřtek

    2 years ago
    Hi all, I am trying to solve the exact same issue: https://prefect-community.slack.com/archives/CL09KU1K7/p1585863863351300 The given solution is not working for me. I want to use Parameter in shell command:
    from prefect import task, Flow, Parameter
    from prefect.tasks.templates import StringFormatter
    from prefect.tasks.shell import ShellTask
    
    
    msg_task = StringFormatter(template='{my_string}')
    
    shell_task = ShellTask()
    
    with Flow("string-template") as flow:
        name = Parameter('name')
    
        msg_output = msg_task(my_string=name)
        shell_output_static = shell_task(command=f"echo Arthur")
        shell_output_dynamic = shell_task(command=f"echo {msg_output}")
    
    
    flow_state = flow.run(name='Marvin')
    
    
    print(flow_state.result[msg_output].result)
    # Marvin
    
    print(flow_state.result[shell_output_static].result)
    # Arthur
    
    print(flow_state.result[shell_output_dynamic].result)
    # Command failed with exit code 2
    nicholas

    nicholas

    2 years ago
    Hi @Jan Feřtek - you could try something like this:
    from prefect import task, Flow, Parameter
    from prefect.tasks.shell import ShellTask
    
    class ParameterizedShellTask(ShellTask):
      def run(self, param):
        self.command = f"echo {param}"
        super(ParameterizedShellTask, self).run()
    
    with Flow("Dynamic Shell Tasks") as flow:
      name = Parameter("name")
      
      shell_output_dynamic = ParameterizedShellTask()(param=name)
    I find that extending the ShellTask and templating the command makes it really powerful for this sort of parameterized flow
    j

    Jan Feřtek

    2 years ago
    Thanks, @nicholas! Do you know why this extended class does not return stdout?
    from prefect import task, Flow, Parameter
    from prefect.tasks.shell import ShellTask
    
    
    class ParameterizedShellTask(ShellTask):
        def run(self, param):
            self.command = f"echo {param}"
            super(ParameterizedShellTask, self).run()
    
    
    basic_shell_task = ShellTask()
    
    
    with Flow("Dynamic Shell Tasks") as flow:
        shell_output_basic = basic_shell_task(command="echo Arthur")
    
        name = Parameter("name")
        shell_output_dynamic = ParameterizedShellTask()(param=name)
    
    
    flow_state = flow.run(name="Marvin")
    
    print(flow_state.result[shell_output_basic].result)
    # Arthur
    
    print(flow_state.result[shell_output_dynamic].result)
    # None
    Oh, I found it:
    class ParameterizedShellTask(ShellTask):
        def run(self, param):
            self.command = f"echo {param}"
            >>>return<<< super(ParameterizedShellTask, self).run()
    🙂