How to pass data between subflows?

Archive
, ,
1

Here is an example showing how to pass data between subflows:

from prefect import task, flow
from prefect import get_run_logger


@task
def get_name():
    return "Marvin 🦆"


@task
def say_hi(user_name: str):
    logger = get_run_logger()
    logger.info("Hello %s!", user_name)


@flow
def hello_world():
    user = get_name()
    say_hi(user)
    return user


@flow
def log_data_from_other_subflow(user_name: str):
    logger = get_run_logger()
    logger.info("Data received from other flow is: %s", user_name)


@flow
def pass_data_between_subflows():
    user = hello_world()
    log_data_from_other_subflow(user)


if __name__ == "__main__":
    pass_data_between_subflows()
Archive: Prefect 2.0 Beta state

To pass data between subflows, you need to call the .result() method twice. Why?

Let’s look at an example:

from prefect import task, flow
from prefect import get_run_logger


@task
def get_name():
    return "Marvin"


@task
def say_hi(user_name: str):
    logger = get_run_logger()
    logger.info("Hello %s!", user_name)


@flow
def hello_world():
    user = get_name()
    say_hi(user)
    return user


@flow
def pass_data_between_subflows():
    subflow_flow_run_state = hello_world()
    subflow_result = subflow_flow_run_state.result().result()
    logger = get_run_logger()
    logger.info("Final subflow result is: %s", subflow_result)


if __name__ == "__main__":
    pass_data_between_subflows()

Calling the subflow hello_world() returns a flow run state. The first .result() returns a flow run state object.

Now, the hello_world() subflow-flow-run returns a task run state from the task get_name() that is assigned to the object user. This user object is a task run state. To get the result of this task run state, you need to call .result() again.

If you run it, you should see the following output:

18:50:12.271 | INFO    | prefect.engine - Created flow run 'lovely-vulture' for flow 'pass-data-between-subflows'
18:50:12.272 | INFO    | Flow run 'lovely-vulture' - Using task runner 'SequentialTaskRunner'
18:50:12.392 | INFO    | Flow run 'lovely-vulture' - Created subflow run 'affable-caterpillar' for flow 'hello-world'
18:50:12.452 | INFO    | Flow run 'affable-caterpillar' - Created task run 'get_name-2744ebee-0' for task 'get_name'
18:50:12.519 | INFO    | Task run 'get_name-2744ebee-0' - Finished in state Completed(None)
18:50:12.553 | INFO    | Flow run 'affable-caterpillar' - Created task run 'say_hi-b457a9f2-0' for task 'say_hi'
18:50:12.576 | INFO    | Task run 'say_hi-b457a9f2-0' - Hello Marvin!
18:50:12.607 | INFO    | Task run 'say_hi-b457a9f2-0' - Finished in state Completed(None)
18:50:12.644 | INFO    | Flow run 'affable-caterpillar' - Finished in state Completed('All states completed.')
18:50:12.644 | INFO    | Flow run 'lovely-vulture' - Final subflow result is: Marvin
18:50:12.665 | INFO    | Flow run 'lovely-vulture' - Finished in state Completed('All states completed.')
3 Likes