create_flow_run(…, task_args={"task_run_name": "…"})

import os
import random
from typing import Dict, List, Any

import prefect
from faker import Faker
from prefect import Flow, task, unmapped
from prefect.backend import FlowRunView
from prefect.core import Parameter
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run


@task
def say_hello(sub_flow_desc: str) -> None:
    <http://prefect.context.logger.info|prefect.context.logger.info>(f"hello from subflow: {sub_flow_desc}")
    if random.randint(0, 10) > 5:
        raise ValueError("fake error")


with Flow("example_sub_flow") as sub_flow:
    sub_flow_desc = Parameter("sub_flow_desc")
    say_hello(sub_flow_desc)


@task
def calculate_sub_flows() -> List[Dict[str, Any]]:
    fake = Faker()
    return [{"sub_flow_desc": fake.name()} for _ in range(10)]


@task
def create_sub_flow(**kwargs) -> Dict[str, str]:
    desc = kwargs["parameters"]["sub_flow_desc"]
    result = create_flow_run.run(**kwargs)
    return {"flow_run_id": result, "sub_flow_desc": desc}


@task
def wait_for_sub_flow(sub_flow: Dict[str, str], **kwargs: Any) -> FlowRunView:
    kwargs.pop("sub_flow_runs", None)
    flow_run_id = sub_flow["flow_run_id"]
    return wait_for_flow_run.run(flow_run_id=flow_run_id, **kwargs)


with Flow("example_parent_flow") as parent_flow:
    sub_flow_params = calculate_sub_flows()

    sub_flow_runs = create_sub_flow.map(
        parameters=sub_flow_params,
        flow_name=unmapped("example_sub_flow"),
        project_name=unmapped("examples"),
        task_args={
            "task_run_name": lambda **kw: f"run subflow: {kw['parameters']['sub_flow_desc']}"
        },
    )

    wait_for_sub_flow.map(
        sub_flow_runs,
        sub_flow_runs=sub_flow_runs,
        raise_final_state=unmapped(True),
        task_args={
            "task_run_name": lambda **kw: f"wait for subflow: {kw['sub_flow_runs']['sub_flow_desc']}"
        },
    )