Hello everyone, I have a question about how to handle failures in flow runs: I have a basic ETL pipeline set up as a flow that is scheduled to execute daily, and it ingests data from about a 100 different endpoints. It is expected that the ingest for most of those endpoints will be successful, but also that there will always be several that are unsuccessful. But since some are unsuccessful, Prefect shows the entire flow run as a failure. Is there a way to change that? I'd prefer that it only show as a failure if none of the ingests are successful, because it's set up to alert people who use this about failures. The Flow object in Python is set up sort of like this

source_ids = ["source_1", "source_2", "source_3"]
with Flow("my_flow") as flow:
    for source_id in source_ids:
        data = extract(source_id)
        data = transform(data)
        load(data)
    return flow

Thanks in advance!

Hi @Alexander Kloumann, that type of control flow is not supported in prefect 1 and so you're not able to get visibility into the state of each task that makes up your flow What you want to use here is mapping, which would look like

source_ids = ["source_1", "source_2", "source_3"]
with Flow("my_flow") as flow:
   data = extract.map(source_ids)
   transformed_data = transform.map(data)
   load.map(transformed_data)

that way, you can respond to mapped task states independently using triggers note that you don't need a

return

statement here, since in prefect 1, flows are not function definitions

I think for Prefect 1.0 I used case to check for flow states and only passed along mapped flows that succeeded. https://docs-v1.prefect.io/api/latest/tasks/control_flow.html#case

Thanks a bunch for the answers @Nate and @Jimmy Le! Sorry for the late reply, I'm not in the rhythm of using Slack lately.