How to customize how the name of a task triggering a child flow run is displayed in the parent DAG? (i.e. how to override the default task name create_flow_run)

Problem

Normally, when you use the create_flow_run task, your child flow run will be displayed in the UI as a task with name create_flow_run, which may be especially unhelpful if you have multiple child flows that you trigger from a parent flow. This example illustrates the issue:

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

FLOW_NAME = "parent_flow_example"


@task(log_stdout=True)
def hello_world():
    print(f"Hello from {FLOW_NAME}!")
    return FLOW_NAME


with Flow(FLOW_NAME) as flow:
    normal_non_subflow_task = hello_world()
    first_child_flow_run_id = create_flow_run(
        flow_name="child_flow_name",
        project_name="PREFECT_PROJECT_NAME",
        upstream_tasks=[normal_non_subflow_task],
    )
    first_child_flowrunview = wait_for_flow_run(
        first_child_flow_run_id, raise_final_state=True, stream_logs=True,
    )

if __name__ == "__main__":
    flow.visualize()

Solution

You can set a custom name using the task_args=dict(name="your_custom_task_name"), argument available on any Prefect task.

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

FLOW_NAME = "parent_flow_example"


@task(log_stdout=True)
def hello_world():
    print(f"Hello from {FLOW_NAME}!")
    return FLOW_NAME


with Flow(FLOW_NAME) as flow:
    normal_non_subflow_task = hello_world()
    first_child_flow_run_id = create_flow_run(
        flow_name="child_flow_name",
        project_name="PREFECT_PROJECT_NAME",
        task_args=dict(name="Name it as you like and this will show in the DAG"),
        upstream_tasks=[normal_non_subflow_task],
    )
    first_child_flowrunview = wait_for_flow_run(
        first_child_flow_run_id,
        raise_final_state=True,
        stream_logs=True,
        task_args=dict(name="Wait for the first subflow"),
    )

if __name__ == "__main__":
    flow.visualize()

Other resources

If you would also like to override the randomly generate flow run name (which by default creates flow run names with a random adjective + animal combination such as “amazing-chicken” :smile: ) for the child flow, you can use the run_name argument on the create_flow_run task.

There is even a way how you can set this name dynamically using a separate task. This topic shows how to do it:

2 Likes