import os

os.environ["PREFECT__LOGGING__LEVEL"] = "DEBUG"

from prefect import Client, Flow, Parameter, task, unmapped
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run


@task
def log_child_task(parameter):
    print(f"child_task(parameter={parameter})")


@task
def log_parent_task(children_parameters):
    print(f"parent_task(children_parameters={children_parameters})")


with Flow("child_flow") as child_flow:
    parameter = Parameter("parameter")
    log_child_task(parameter)


with Flow("parent_flow_failing") as parent_flow_failing:
    children_parameters = Parameter("children_parameters")

    a = log_parent_task(children_parameters)

    # Run a child_flow for every dict of parameters
    child_run_ids = create_flow_run.map(
        parameters=children_parameters,
        flow_name=unmapped("child_flow"),
        upstream_tasks=[a],
    )
    flow_runs_finished = wait_for_flow_run.map(flow_run_id=child_run_ids)


with Flow("parent_flow_succeeding") as parent_flow_succeeding:
    children_parameters = Parameter("children_parameters")

    a = log_parent_task(children_parameters)

    # Same, but only one child_flow
    child_run_id = create_flow_run(
        parameters=children_parameters[0],
        flow_name="child_flow",
        upstream_tasks=[a],
    )
    flow_run_finished = wait_for_flow_run(flow_run_id=child_run_id)


def run_parent_flow():
    # Register the flows so that they can be found on the server:
    project_name = "test"
    Client().create_project(project_name=project_name)
    parent_flow_failing.register(project_name=project_name)
    child_flow.register(project_name=project_name)
    parent_flow_succeeding.register(project_name=project_name)

    print("A flow using create_flow_run.map")
    state = parent_flow_failing.run(
        parameters={
            "children_parameters": [
                {"parameter": 1},
                {"parameter": 2},
            ]
        },
    )

    print("\nA flow using create_flow_run without map")
    state = parent_flow_succeeding.run(
        parameters={
            "children_parameters": [
                {"parameter": 1},
                {"parameter": 2},
            ]
        },
    )


if __name__ == "__main__":
    run_parent_flow()

[2021-09-22 10:55:32+0200] INFO - prefect.TaskRunner | Task 'create_flow_run': Starting task run...
[2021-09-22 10:55:32+0200] DEBUG - prefect.TaskRunner | Task 'create_flow_run': Handling state change from Pending to Failed
[2021-09-22 10:55:32+0200] INFO - prefect.TaskRunner | Task 'create_flow_run': Finished task run for task with final state: 'Failed'