https://prefect.io logo
Powered by
Title
s

sark

09/16/2020, 3:01 AM
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
?
n

nicholas

09/16/2020, 3:09 AM
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
s

sark

09/16/2020, 3:12 AM
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")
?
n

nicholas

09/16/2020, 3:14 AM
Hm, good question; I think that'll create a Constant task as a placeholder to do that interpolation.
s

sark

09/16/2020, 3:16 AM
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')
n

nicholas

09/16/2020, 3:30 AM
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
s

sark

09/16/2020, 3:33 AM
ah when i do 
CreateContainer()(image_name=image_name)
it does work in that all the tasks are green
n

nicholas

09/16/2020, 3:33 AM
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?
s

sark

09/16/2020, 3:35 AM
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!
n

nicholas

09/16/2020, 3:36 AM
Awesome, glad to help! Sorry for the confusion earlier, writing code at midnight is hazardous 😬
s

sark

09/16/2020, 3:37 AM
no worries, i am UTC+8 over here 🙂
🌏 1
n

nicholas

09/16/2020, 3:39 AM
Hopefully your code writing goes better than mine at UTC-5! 😆
#prefect-communityJoin Slack
Powered by