Greetings, I'm having an issue with passing a Parameter from the command line to my flow, if I'm correct on the server the Parameter can be passed in the UI, but for testing I need to pass in sys.argv[2]. here's the relevant code:

def register__flow():

    

with Flow("FlowName", schedule=None, ) as flow:

        

file_name = Parameter(name='file_name',default=None)

        

params = get_params()

        

p = get_file_blob(

                        

file_name=f"notebooks/{file_name}",

                        

con_string=params["uploads_blob"],

                        

container="uploads"

        

)

        

flow.add_edge(file_name,params)

        

flow.add_edge(params, p)

    

if len(sys.argv) > 1 and sys.argv[1] == "register":

        

flow.register(project_name="project_x")

    

elif len(sys.argv) > 1 and sys.argv[1] == "test" and len(sys.argv[2]) > 1:

        

flow.run(parameters={'file_name':sys.argv[2]})

    

else:

        

print("Please use a quoted string of the file name...")

if __name__ == "__main__":

    

register__flow()

Thanks

Hi @Frederick Thomas! What error do you get?

I'm getting a file not found error from

Ok I’ll try this on my end and get back to you

Thanks!!! This is driving me nuts...

It works for me. Here is my sample flow:

import prefect
from prefect import Flow, task, Parameter
import sys

@task
def test(x):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(x)
    return

with Flow("test") as flow:
    param = Parameter("param", default= sys.argv[2])
    test(param)
flow.run()

Hmmm, give me a few minutes

Is

get_file_blob

there a task?

@Kevin Kho It is now working as expected:

p = get_file_blob(
                        file_name = task(lambda p: f"notebooks/{p}")(file_name),
                        con_string=params["fileuploader_blob"],
                        container="fu-uploads"
        )

We're currently using version 0.14.12, is there any documentation that is would emphasize this requirement? Otherwise, thanks!

This is a core Prefect thing that

tasks

are used to defer execution. Maybe this documentation on tasks will help you