Hi all, I am trying to solve the exact same issue:...
# prefect-community
j
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:
Copy code
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
👀 1
n
Hi @Jan Feƙtek - you could try something like this:
Copy code
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)
đŸ€© 1
I find that extending the ShellTask and templating the command makes it really powerful for this sort of parameterized flow
j
Thanks, @nicholas! Do you know why this extended class does not return stdout?
Copy code
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:
Copy code
class ParameterizedShellTask(ShellTask):
    def run(self, param):
        self.command = f"echo {param}"
        >>>return<<< super(ParameterizedShellTask, self).run()
🙂