@Marvin Is there a way to start a subflow by the name of the subflow instead of as a function call? I have a deployed main flow, I want to start several subflows from a list of names like

["extract", "create_report", "upload"]

. I don't want to define each subflow as a deployment.

To start a subflow by its name instead of using a function call, you have a couple of options in Prefect 3.x, though they may require some creativity as Prefect primarily expects function calls for subflows. 1. Dynamic Flow Execution: While Prefect doesn't natively support executing a flow from a string name directly, you can create a mapping of flow names to their corresponding flow functions and dynamically execute them. Here's a basic example of how you might achieve this:

python
    from prefect import flow

    # Define your subflows
    @flow
    def extract():
        print("Extracting data...")

    @flow
    def create_report():
        print("Creating report...")

    @flow
    def upload():
        print("Uploading data...")

    # Map subflow names to functions
    subflows = {
        "extract": extract,
        "create_report": create_report,
        "upload": upload
    }

    # List of subflow names to execute
    subflow_names = ["extract", "create_report", "upload"]

    # Execute subflows by name
    for name in subflow_names:
        subflows[name]()

2. Flow Engine: The Prefect flow engine does have some internal methods like

load_subflow_run

and

create_flow_run

that interact with subflows. However, these are not typically exposed for direct use without deploying flows. For more details, you can explore the Prefect flow engine source code. If you're looking for a more native solution or feature, it might involve contributing a feature request or checking for updates in the Prefect repository.

@Marvin I'm interested in the dynamic flow execution example. Can you show the example the subflows also receiving parameters? For example, the

extract()

subflow needs the parameters

file_name: str

and

datastore_type: str

.

Certainly! You can modify the dynamic flow execution example to pass parameters to the subflows. Here's how you might do it:

python
from prefect import flow

# Define your subflows with parameters
@flow
def extract(file_name: str, datastore_type: str):
    print(f"Extracting data from {file_name} using {datastore_type} datastore...")

@flow
def create_report():
    print("Creating report...")

@flow
def upload():
    print("Uploading data...")

# Map subflow names to functions
subflows = {
    "extract": extract,
    "create_report": create_report,
    "upload": upload
}

# List of subflow names to execute with their respective parameters
subflow_params = {
    "extract": {"file_name": "data.csv", "datastore_type": "SQL"},
    "create_report": {},
    "upload": {}
}

# Execute subflows by name with parameters
for name, params in subflow_params.items():
    subflows<**params|name>

In this example, the

extract

subflow is executed with the parameters

file_name

and

datastore_type

. Each subflow can have its own set of parameters defined in the

subflow_params

dictionary. This approach allows you to flexibly manage and execute various subflows with their specific parameters.

@Marvin Got it -- so this is making use of python treatment of functions as first class objects. The key is storing the function name in a dictionary with key being a string of the function name. This dictionary value is callable. Cool.

Exactly. By leveraging Python's ability to treat functions as first-class objects, you can store them in a dictionary and call them dynamically using string keys. This approach provides flexibility in managing and executing different functions based on runtime conditions, which is particularly useful in scenarios like yours where you want to dynamically trigger subflows. If you have any more questions or need further assistance, feel free to ask.