sark
09/16/2020, 3:01 AMCreateContainer
?nicholas
09/16/2020, 3:09 AMwith Flow("Some Flow") as flow:
image_name = Parameter("image_name")
create = CreateContainer()(image_name=image_name)
which will create a data dependency between the twosark
09/16/2020, 3:12 AMCreateContainer()(image_name=f"myimage:{version})
where version = Parameter("version")
?nicholas
09/16/2020, 3:14 AMsark
09/16/2020, 3:16 AMCreateContainer(image_name=image_name)
?@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
09/16/2020, 3:30 AMIn [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)
sark
09/16/2020, 3:33 AMCreateContainer()(image_name=image_name)
it does work in that all the tasks are greennicholas
09/16/2020, 3:33 AMCreateContainer()
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 liststart = StartContainer()
, by referencing start()
in your flow contextsark
09/16/2020, 3:35 AMCreateContainer()(image_name=image_name)
instead of CreateContainer(image_name=image_name)
nicholas
09/16/2020, 3:36 AMsark
09/16/2020, 3:37 AMnicholas
09/16/2020, 3:39 AM