p

    Philip MacMenamin

    2 years ago
    I have a question about using
    Parameter
    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 args
    Chris White

    Chris White

    2 years ago
    Hi @Philip MacMenamin - you should treat Parameters like normal tasks. So for example:
    from 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")
    For the shell task specifically, I suggest customizing it to your needs via subclassing:
    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 way
    p

    Philip MacMenamin

    2 years ago
    ok - so for the first question, you always pass the param to a task for it to be used. You never use it within the flow block.
    And, ok, subclass out the shellTask class to call it with args. Got it. Thanks!
    class MvTask(ShellTask):
    def run(self, custom_param):
    # do stuff with custom_param
    return super().run(command = f"util_name {custom_param}")
    Chris White

    Chris White

    2 years ago
    yup that looks good to me
    p

    Philip MacMenamin

    2 years ago
    so if I wanted to pass a string to the shell task, I set the command this way?
    ok, thanks