Manuel Aristarán
04/02/2020, 9:44 PMParameter
. My use case is building a shell command:
data_source_id = Parameter("data_source_id")
# ...
t = shell_task(command=f"some_script {data_source_id}")
Jeremiah
04/02/2020, 9:44 PMParameter
as the input to a different task. In your case, you might need to use a StringFormatter
(see docs) to have a task to pass the value to:
# create the task
fmt = StringFormatter(template='some_script{data_source_id}')
with Flow("test") as flow:
did = Parameter('did')
# pass the parameter to the string formatter
string = fmt(data_source_id=did)
# pass the string to the shell task
shell_task(cmd=string)
Parameter
to any other task, it’s value becomes the keyword argument input to that task)prefect.context.get('parameters')
, so you can template the string with that context value. However, you need to make sure you apply the string template at runtime and not task initialization.Task
because that’s the only environment that has access to the Parameter value.Manuel Aristarán
04/02/2020, 9:51 PMwith Flow("") as f:
shell_task(cmd=f"some_script {prefect.context.get('parameters')['data_source_id']}")
Jeremiah
04/02/2020, 9:52 PMprefect.context()
isn’t populated until you call flow.run()
, so that code will fail to load the parameter valueManuel Aristarán
04/02/2020, 9:52 PMJeremiah
04/02/2020, 9:52 PMStringFormatter
’s job is to take what you did there (which would happen during flow definition) and make it possible to do at runtime insteadManuel Aristarán
04/02/2020, 9:52 PMJeremiah
04/02/2020, 9:54 PMManuel Aristarán
04/02/2020, 9:58 PMJan Feřtek
07/24/2020, 5:36 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, 5:39 PMJan Feřtek
07/24/2020, 5:57 PM