Hello! I'm working to port an Argo Workflow to Prefect. All of our Argo WFs are docker based, as such, I'm creating a simple flow that takes 2 Parameters and computes the docker command to execute. The problem I'm running into is that by building the DAG correctly I am referencing a global variable technically before it is defined. This is not a pretty situation, however it is technically producing the correct DAG (based on dag visualization and execution) I've included a paired down snippet. What is the correct/preferred way to accomplish a task like this? Essentially I need to set a Task's arguments based on the execution of a prior Task.

...

cft_container = CreateContainer(image_name="twodub:v0.4")
cft_start = StartContainer()
cft_logs = GetContainerLogs()
cft_status_code = WaitOnContainer()


@task
def set_cft_container_command(cft_name: str, account_id: str):
    logger = prefect.context.get("logger")
    ret = f"python -m twodub cft install --cft-name {cft_name} --account-id {account_id}"
    new_container.command = ret # <--- Referencing new_container before declared
    <http://logger.info|logger.info>(f"command is: '{ret}'")
    return ret


with Flow("install-cft-flow") as flow:
    cft_name = Parameter("CFT Name")
    account_id = Parameter("Account ID")

    scc = set_cft_container_command(cft_name, account_id)

    # I'm creating a new version of this Task here with bind
    # I suspect there is a cleaner way than this, but it wasn't
    # apparent to me in the documentation 
    #
    # If I did not do this, then the container would be created
    # before the set_cft_container_command was called and the
    # correct command argument would not be set
    new_container = cft_container().bind(upstream_tasks=[scc])

    start_container = cft_start(container_id=new_container)

...

Hi @Ken Farr, you can do

new_container = CreateContainer(scc)

instead. You can initialize tasks inside the Flow context.

What I really want is to just set the argument

command

based on the two Parameters being passed in. If you take a peek at

set_cft_container_command

the only purpose of this task is to build the container command and set it. BUT since the run() on CreateContainer doesn't take a

command

I have to do it on the Task instead. It's creating a weird circular logic that I think can be fixed by expanding the CreateImage API to accept the

command

at execution instead of at creation

This is from the Task Library right? The CreateContainer seems to take a

command

in the run method

Yep, that's what I'm following. But CreateContainer runs not when the flow runs but when the flow is created. I need to set the

command

when the flow executes, not when the flow is created. Such is my understanding.

The init happens during build time, and the run happens during execution time. You can do:

with Flow("install-cft-flow") as flow:
    cft_name = Parameter("CFT Name")
    account_id = Parameter("Account ID")
    scc = set_cft_container_command(cft_name, account_id)
    CreateContainer()(image_name="twodub:v0.4", command=scc)

Yes! That was it.

No problem! Just a note that not all tasks in the task library have everything available to configure at runtime. This is something on the radar to fix.