Jack Sundberg

    Jack Sundberg

    3 months ago
    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()
    Kevin Kho

    Kevin Kho

    3 months ago
    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.
    Jack Sundberg

    Jack Sundberg

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

    Anna Geller

    3 months ago
    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
    Jack Sundberg

    Jack Sundberg

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

    Anna Geller

    3 months ago
    Yup exactly