Hello! Is there any easy way to pass result of one task (a dict) into the input of another (as **kwargs)? Here's a minimal example of what I'd like to do if it helps:

import prefect
from prefect import task, Flow

@task
def parse_task():
    name = prefect.context.flow_name
    return {"name": name, "test": 123}


@task
def split_kwargs_task(name, test):
    print(name)
    print(test)


# Does NOT work, but desired approach
with Flow("hello-flow") as flow:
    parameters_cleaned = parse_task()
    split_kwargs_task(**parameters_cleaned)  # <-- raises error

# Manually pulling out each kwarg gets this to work
with Flow("hello-flow") as flow:
    parameters_cleaned = parse_task()
    split_kwargs_task(
        name=parameters_cleaned["name"],
        test=parameters_cleaned["test"],
    )

flow.run()

Not as kwargs because the

**

operator is evaluated during DAG construction time, not Flow run time. There is magic here

parameters_cleaned["name"]

happens during runtime because it’s a task. There is a GetItem task that is added to the flow when you use this syntax on a Task object.

Makes sense -- I can see those tasks when I visualize the flow. Do you know if this is possible with prefect 2.0?

yup, it will get easier in Prefect 2.0 since it's just python in 1.0 you can use

task_args

to pass some extra task args

Awesome! For the short-term while I'm using 1.0, does task_args take a dictionary?

Yup exactly