Hey everyone! I’m trying to automate API calls using the map functionality. are the conditional tasks able to work with the mapping results or do I need to reduce the results before performing the condition

perfect should be able to access the result of the map task when you use it in a call, you do not need to explicitly call reduce https://docs.prefect.io/core/concepts/mapping.html#reduce

is there an easy way to collect task state data in the pipeline for report. for example, 8/10 map tasks were successful

Hey @John Ramirez, nothing is immediately coming to mind but I’ll respond in a bit

Definitely! You can use this query directly:

query {
  mapped_children(task_run_id: "<<task run id>>") {
    max_end_time
    min_start_time
    state_counts
  }
}

which will give a lot of good info about the mapping

You can do something like this to check if the tasks returned Exceptions. Note that you need an

all_finished

trigger to make sure the downstream task runs

import random
import prefect
from prefect import Flow, task


@task
def fail_sometimes(i):
    # fail
    if i == 2:
        # We can either raise a signal
        raise prefect.engine.signals.FAIL("FAILED")

    else:
        return i

# The default trigger is 'all_successful' and we need to allow failures upstream for
# this task to handle them robustly
@task(trigger=prefect.triggers.all_finished)
def check_what_failed(results):
    failures = len([x for x in results if isinstance(x, Exception)])
    successes = len(results) - len(failures)
    return 

with Flow(
    "mapped-failures",
) as flow:
    map_results = fail_sometimes.map([1,2,3,4])
    check_what_failed(map_results)

flow.run()

thanks!

I have never seen pytest run as code. What are validating on Pandas dataframes?