Probably a stupid question: how do I get the value...
# prefect-community
m
Probably a stupid question: how do I get the value of a
Parameter
. My use case is building a shell command:
Copy code
data_source_id = Parameter("data_source_id")
# ...
t = shell_task(command=f"some_script {data_source_id}")
j
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:
Copy code
# 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
Right. So would this work?
Copy code
with Flow("") as f:
  shell_task(cmd=f"some_script {prefect.context.get('parameters')['data_source_id']}")
(yea, ugly :))
j
No, unfortunately -
prefect.context()
isn’t populated until you call
flow.run()
, so that code will fail to load the parameter value
m
Perfect. I’ll go with the template 🙂
j
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
👌
m
Thanks a lot!
👍 1
j
Let us know if that works!
m
it did 🙂 thanks again
j
Hi, I can't get it working. What am I missing?
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
n
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