@Marvin how do I run a list of flow objects as subflows within the current flow and then collect the results so I can say which ones succeeded and which ones failed and which ones succeeded?

You want to run multiple flows as subflows within a parent flow and collect their results to determine which ones succeeded and which ones failed. Let me search the Prefect documentation and gather relevant code examples for this use case. To run a list of flow objects as subflows within a parent flow and then collect their results to determine which ones succeeded and which ones failed, you can use the following approach: 1. Define the Parent Flow: This is the main flow that will call the other flows as subflows. 2. Define the Subflows: These are the flows you want to run as subflows. 3. Run Subflows in a Loop: In the parent flow, run each subflow within a loop and collect the results. 4. Check Status of Each Subflow: After running each subflow, check its state to determine if it succeeded or failed. Here's a code example to illustrate this:

python
from prefect import flow, task
from prefect.engine.results import State

# Define a sample subflow 
@flow
def subflow(index: int):
    if index % 2 == 0:
        return f"Subflow {index} - Succeeded"
    else:
        raise Exception(f"Subflow {index} - Failed")

# Define the parent flow
@flow
def parent_flow(subflow_count: int):
    results = {}
    for i in range(subflow_count):
        try:
            # Run the subflow
            subflow_result = subflow.with_options(name=f"Subflow {i}")(i)
            results[i] = (subflow_result, "Succeeded")
        except Exception as e:
            results[i] = (str(e), "Failed")
    
    # Print the results
    for i, (result, status) in results.items():
        print(f"Subflow {i}: {status} - {result}")

# Run the parent flow with a specified number of subflows
parent_flow(5)

Explanation: - Subflows: Defined using the

@flow

decorator. The example subflow raises an exception for odd-indexed runs to simulate failure. - Parent Flow: Runs each subflow and captures the result or exception. The results are stored in a dictionary with their status for later evaluation. You can expand on this example to cover more complex cases or use Prefect’s features to manage state and logging more effectively. For more details, you can refer to the Prefect documentation: - Concepts: Flows: https://docs.prefect.io/latest/concepts/flows/ - Composing Flows Example: https://raw.githubusercontent.com/zzstoatzz/prefect-code-examples/main/examples/flows/basic.py Feel free to modify the code as needed to match your specific use case.

@Marvin So there's no option to simply return the result regardless of whether it succeeded or failed, in other words, to eliminate the need for a try statement?

You can eliminate the need for a try-except block by using the

return_state=True

and

state.result(raise_on_failure=False)

functionalities in Prefect. This will allow you to handle the state of each subflow run directly. Here's an example of how you can achieve this:

python
from prefect import flow, task
from prefect.engine.results import State

# Define a sample subflow 
@flow
def subflow(index: int):
    if index % 2 == 0:
        return f"Subflow {index} - Succeeded"
    else:
        raise Exception(f"Subflow {index} - Failed")

# Define the parent flow
@flow
def parent_flow(subflow_count: int):
    results = {}
    for i in range(subflow_count):
        subflow_state = subflow.with_options(name=f"Subflow {i}").run(i, return_state=True)
        subflow_result = subflow_state.result(raise_on_failure=False)

        if subflow_state.is_failed():
            results[i] = (subflow_result, "Failed")
        else:
            results[i] = (subflow_result, "Succeeded")
    
    # Print the results
    for i, (result, status) in results.items():
        print(f"Subflow {i}: {status} - {result}")

# Run the parent flow with a specified number of subflows
parent_flow(5)

Explanation: - Subflows: Defined using the

@flow

decorator. The example subflow raises an exception for odd-indexed runs to simulate failure. - Parent Flow: Runs each subflow and captures the state directly using

return_state=True

. - Uses

state.result(raise_on_failure=False)

to get the result without raising an exception. - Checks the state using

subflow_state.is_failed()

to determine if the subflow failed or succeeded. - Stores the results and their status (succeeded or failed) in a dictionary. This approach makes it simpler to handle the results of subflows without using a try-except block. For more details, you can refer to the Prefect documentation: - Concepts: Results: https://docs.prefect.io/latest/concepts/results/ Feel free to modify the code as needed to match your specific use case.

@Marvin what kind of object is returned by state.result(raise_on_failure=False)? A list? How can I get a human readable subflow result?