Thread
#prefect-community
    sark

    sark

    2 years ago
    hi guys from the docs i know how to pass a parameter to a new task i define myself, but how do i pass parameters to an existing task like
    CreateContainer
    ?
    nicholas

    nicholas

    2 years ago
    Hi @sark - you should be able to pass params to existing tasks in the same way, example:
    with Flow("Some Flow") as flow:
      image_name = Parameter("image_name")
    
       create = CreateContainer()(image_name=image_name)
    which will create a data dependency between the two
    sark

    sark

    2 years ago
    thanks nicholas, does it also work when i don’t pass in the parameter as a whole argument but as an interpolation like with
    CreateContainer()(image_name=f"myimage:{version})
    where
    version = Parameter("version")
    ?
    nicholas

    nicholas

    2 years ago
    Hm, good question; I think that'll create a Constant task as a placeholder to do that interpolation.
    sark

    sark

    2 years ago
    ok let me try it
    hmm now the parameter tasks don’t get created anymore
    btw why was it CreateContainer()(image_name=image_name)and not just
    CreateContainer(image_name=image_name)
    ?
    ok i defined an intermediate task for preparing the string
    @task
    def image(version):
       return f"myimage:{version}"
    
    start = StartContainer()
    
    with Flow("flow", storage=GCS('bucket')) as flow:
        flow.environment = LocalEnvironment(labels=["docker"])
    
        version = Parameter('version', required=True)
    
        v = image(version)
        c = CreateContainer(image_name=v)
        start_container = start(c)
    i get
    Unexpected error: TypeError('Object of type FunctionTask is not JSON serializable')
    nicholas

    nicholas

    2 years ago
    Ah, that's cause I gave you bad code, sorry about hat, long day over here 🤦
    You can you something like this to create your image name:
    In [12]: @task
        ...: def create_image_name(version):
        ...:     return f"myimage:{version}"
        ...:
    
    In [13]: with Flow("Some Flow") as flow:
        ...:     image_name_param = Parameter("image_name")
        ...:     image_name = create_image_name(version=image_name_param)
        ...:     create = CreateContainer()(image_name=image_name)
    Oops, you've already got that part
    sark

    sark

    2 years ago
    ah when i do
    CreateContainer()(image_name=image_name)
    it does work in that all the tasks are green
    nicholas

    nicholas

    2 years ago
    CreateContainer()
    instantiates an instance of the task class, adding
    ()
    ensures the class's default run method is called, meaning it'll be added to the flow task list
    You've done the same thing with
    start = StartContainer()
    , by referencing
    start()
    in your flow context
    Did you resolve the JSON serialization issue?
    sark

    sark

    2 years ago
    ah i see
    yes i resolved it by doing
    CreateContainer()(image_name=image_name)
    instead of
    CreateContainer(image_name=image_name)
    and the interpolation is correct!
    thanks for your help!
    nicholas

    nicholas

    2 years ago
    Awesome, glad to help! Sorry for the confusion earlier, writing code at midnight is hazardous 😬
    sark

    sark

    2 years ago
    no worries, i am UTC+8 over here 🙂
    nicholas

    nicholas

    2 years ago
    Hopefully your code writing goes better than mine at UTC-5! 😆