Philip MacMenamin
06/29/2020, 7:01 PMParameter
in ShellTask
According to the docs, Parameter
is a type of Task
, and I'm guessing it needs to get the method run
called on it before it's usable. I'm guessing that this method is called when you pass the parameter to a function with the @task
decorator?
I have a shell task, which I'd like to pass an object (or a string) which is created using a Parameter
object. Normally Shell tasks seem to be defined in the flow block. When I try to run a shell task, in a flow, with arguments that are dependent on a parameter, it fails with something like:
Traceback (most recent call last):
File "/usr/lib/python3.8/runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "/usr/lib/python3.8/runpy.py", line 86, in _run_code
exec(code, run_globals)
File "/home/macmenaminpe/code/prefect/pdb_flow/parameterized_flow.py", line 45, in <module>
num_lines = s_task(command=f"wc -l {job.pdb_fp} > {job.job_dir}num_lines.txt")
File "/home/macmenaminpe/.local/lib/python3.8/site-packages/prefect/tasks/core/function.py", line 68, in __getattr__
raise AttributeError(f"'FunctionTask' object has no attribute {k}")
AttributeError: 'FunctionTask' object has no attribute pdb_fp
So, I guess the two questions are:
• how do I "get at" the input val of a Parameter in a flow (or do I not do this, and always pass that param out to a @task
and let it get handled there
• how do I create a shellTask that takes argsChris White
06/29/2020, 7:08 PMfrom prefect import task, Parameter
@task
def my_parametrized_task(p):
print(p)
with Flow("example") as flow:
param = Parameter("p", default=42)
my_parametrized_task(param)
flow.run()
flow.run(p="new value")
class MyTask(ShellTask):
def run(self, custom_param):
# do stuff with custom_param
return super().run(**kwargs) # pass whatever kwargs you want to the parent run method
and then add this to your flow in the standard wayPhilip MacMenamin
06/29/2020, 7:11 PMclass MvTask(ShellTask):
def run(self, custom_param):
# do stuff with custom_param
return super().run(command = f"util_name {custom_param}")
Chris White
06/29/2020, 7:20 PMPhilip MacMenamin
06/29/2020, 7:21 PM