map_task_mre.py

from typing import List
from flytekit import map_task, task, workflow


@task
def my_task(should_succeed: bool) -> bool:
    if should_succeed:
        return should_succeed
    raise ValueError("This is a failure")


@task
def get_num_successes_task(my_task_results: List[bool]) -> int:
    return len(my_task_results)


@workflow
def my_map_task_workflow(inputs: List[bool]) -> int:
    my_task_results = map_task(
        my_task,
        min_success_ratio=0.5,
    )(should_succeed=inputs)
    num_successes = get_num_successes_task(my_task_results=my_task_results)
    return num_successes

num_successes = my_map_task_workflow(inputs=[True, True, False])