How can I create a subflow?

Prefect 2.0

In order to create a subflow, decorate your subflow function with @flow and call it within your parent flow. If you want to retrieve data returned by the subflow, use .result():

from prefect import flow

def subflow():
    return "Hello from subflow!"

def main_flow():
    subflow_state = subflow()

if __name__ == "__main__":
    main_flow_state = main_flow()

Prefect 1.0

Prefect 1.0 leverages separate tasks to create a child flow run within a parent flow. The create_flow_run_task and StartFlowRun tasks invoke a child flow run and return their flow_run_id. Migrating those to Orion is as easy as calling your child flows from a parent flow.

from prefect import Flow
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run

PROJECT_NAME = "community"

with Flow("sample_flow") as flow:
    child_run_id = create_flow_run(
        task_args={"name": "Friendly name for a DAG node"},
    extract_load_wait_task = wait_for_flow_run(
        task_args={"name": "Friendly wait task name"},

:point_right:t2: Technically speaking, the concept of a subflow in its true sense does not exist in Prefect 1.0. The real first-class subflow feature was introduced in Orion. Prefect 1.0, however. provides the orchestrator pattern allowing to trigger child flows from parent flows via an API call.