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
Jeremiah
Parameter
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)
Jeremiah
Parameter
to any other task, it’s value becomes the keyword argument input to that task)Jeremiah
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.Jeremiah
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']}")
Manuel Aristarán
04/02/2020, 9:51 PMJeremiah
prefect.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
StringFormatter
’s job is to take what you did there (which would happen during flow definition) and make it possible to do at runtime insteadJeremiah
Manuel Aristarán
04/02/2020, 9:52 PMJeremiah
Manuel 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
Jan Feřtek
07/24/2020, 5:57 PM