Jan Feřtek
07/24/2020, 5:57 PMfrom 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
07/24/2020, 6:06 PMfrom 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)
Jan Feřtek
07/24/2020, 8:08 PMfrom 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
class ParameterizedShellTask(ShellTask):
def run(self, param):
self.command = f"echo {param}"
>>>return<<< super(ParameterizedShellTask, self).run()
🙂