How to dynamically change the run_name in the create_flow_run task?

Archive
, , , , , ,
1

Currently, it is not possible to pass a parameter value to a run_name using an f-string, as shown in the following example.

Child flow:

from prefect import task, Flow, Parameter


@task(log_stdout=True)
def hello_world(name_: str = "world"):
    print(f"hello {name_}")


with Flow("01_child_flow") as flow:
    param = Parameter("user_name", default="Marvin")
    hw = hello_world(param)

Parent flow:

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

with Flow("02_parent_flow") as parent_flow:
    param = Parameter("user_name", default="world")
    flow_run_id = create_flow_run(
        flow_name="01_child_flow",
        project_name="xyz",
        parameters=dict(user_name=param),
        run_name=f"API {param}",
    )

or:

from prefect import Flow, Parameter
from prefect.tasks.prefect import StartFlowRun


child_flow = StartFlowRun(flow_name="01_child_flow", project_name="xyz",)


with Flow("02_parent_flow") as parent_flow:
    param = Parameter("user_name", default="world")
    flow_run_id = child_flow(parameters=dict(user_name=param), run_name=f"API {param}",)

Each of those parent flow runs will result in the child flow run with the following name:

image
image1198×252 26 KB

However, using only the parameter value (without an f-string) as run_name works fine:

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

with Flow("02_parent_flow") as parent_flow:
    param = Parameter("user_name", default="world")
    flow_run_id = create_flow_run(
        flow_name="01_child_flow",
        project_name="xyz",
        parameters=dict(user_name=param),
        run_name=param,
    )

image
image1228×300 21.4 KB

Solution

To solve it, use a separate task that generates and returns the desired run_name.

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


@task
def get_curom_run_name(param: str):
    return f"API {param}"


with Flow("02_parent_flow") as parent_flow:
    param = Parameter("user_name", default="world")
    child_flow_run_name = get_curom_run_name(param)
    flow_run_id = create_flow_run(
        flow_name="01_child_flow",
        project_name="xyz",
        parameters=dict(user_name=param),
        run_name=child_flow_run_name,
    )