Is there a way to map

StartFlowRun

? I've got a task

spawn_time_series_import = StartFlowRun(
        project_name="python_imports", 
        flow_name="blast_metric_series_flow")

which takes an argument

spawn = spawn_time_series_import(
                parameters=dict(action_id=blast_id))

and a task that returns a list of `blast_id`s with which I want to spawn many flows

blast_id_list = push_records_to_summary_table(imported)

I want to put all this together in something like this, but I'm having trouble finding the right syntax

spawn_time_series_import = StartFlowRun(
        project_name="python_imports", 
        flow_name="blast_metric_series_flow")



with Flow("blast_import_flow",
        schedule=daily_schedule,
        state_handlers=[cloud_only_slack_handler]
        ) as flow:

    [[[stuff]]]
    blast_id_list = push_records_to_summary_table(imported)
    spawn = spawn_time_series_import.map(
                parameters=dict(action_id=blast_id_list))

is there an accepted syntax for this?

Hi @Chris Jordan - I think this explanation and example should answer your question: https://prefect-community.slack.com/archives/CL09KU1K7/p1607727047066800 Let us know if you've any other questions.

Thanks for the response @Jenny. I'm trying to do something similar to this, but not quite the same. In this case I'm not interested in getting the results out of the flow run I started, as OP in that thread is. The challenge I'm struggling with is that I have a task result that contains a list of N items, and I want to feed that to N instances of

StartFlowRun

. If I had used a task instead of a flow for this, I could use map (and that is my fallback plan), but I would prefer to have each of these be their own flow run if possible.

It looks like your issue is that you’re trying to pass the mapped argument in a dict which as a simple example looks like — this won’t work. I’ll create a passing example in a second.

import prefect
from prefect import task, Flow


tasks = [i for i in range(3)]


@task
def display(params):
    <http://prefect.context.logger.info|prefect.context.logger.info>(f"{params['i']}")


with Flow("mapped") as flow:
    display.map(params=dict(i=tasks))

flow.run()

I'll see what I can do with that - thank you @Zanie