https://prefect.io logo
Title
m

Manuel Aristarán

04/02/2020, 9:44 PM
Probably a stupid question: how do I get the value of a
Parameter
. 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}")
j

Jeremiah

04/02/2020, 9:44 PM
Not a stupid question 🙂 There are two ways
The first is to use the
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)
(Sorry I didn’t test that code but the rough idea is if you pass a
Parameter
to any other task, it’s value becomes the keyword argument input to that task)
The second way is that at runtime, the parameter values are available in
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.
Both ways differ from your current implementation in that the string interpolation must take place at runtime inside a
Task
because that’s the only environment that has access to the Parameter value.
m

Manuel Aristarán

04/02/2020, 9:51 PM
Right. So would this work?
with Flow("") as f:
  shell_task(cmd=f"some_script {prefect.context.get('parameters')['data_source_id']}")
(yea, ugly :))
j

Jeremiah

04/02/2020, 9:52 PM
No, unfortunately -
prefect.context()
isn’t populated until you call
flow.run()
, so that code will fail to load the parameter value
m

Manuel Aristarán

04/02/2020, 9:52 PM
Perfect. I’ll go with the template 🙂
j

Jeremiah

04/02/2020, 9:52 PM
The
StringFormatter
’s job is to take what you did there (which would happen during flow definition) and make it possible to do at runtime instead
:yes:
m

Manuel Aristarán

04/02/2020, 9:52 PM
Thanks a lot!
👍 1
j

Jeremiah

04/02/2020, 9:54 PM
Let us know if that works!
m

Manuel Aristarán

04/02/2020, 9:58 PM
it did 🙂 thanks again
j

Jan Feřtek

07/24/2020, 5:36 PM
Hi, I can't get it working. What am I missing?
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
n

nicholas

07/24/2020, 5:39 PM
Hi @Jan Feřtek - would you mind starting a new thread so we don't send alerts to people that don't need them? Thanks 🙏
j

Jan Feřtek

07/24/2020, 5:57 PM