https://prefect.io logo
Join Slack
Powered by
hi guys from the docs i know how to pass a paramet...
# prefect-community
s
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
Hi @sark - you should be able to pass params to existing tasks in the same way, example:
Copy code
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
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
Hm, good question; I think that'll create a Constant task as a placeholder to do that interpolation.
s
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
Copy code
@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
Copy code
Unexpected error: TypeError('Object of type FunctionTask is not JSON serializable')
n
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:
Copy code
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
ah when i do 
CreateContainer()(image_name=image_name)
it does work in that all the tasks are green
n
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
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
Awesome, glad to help! Sorry for the confusion earlier, writing code at midnight is hazardous 😬
s
no worries, i am UTC+8 over here 🙂
🌏 1
n
Hopefully your code writing goes better than mine at UTC-5! 😆
Open in Slack
PreviousNext
Powered by