#flow.py

exists_items = wait_list_creation.map(some_list)
used_upstream_data = use_upsteam_data(
        input_data,
        exists_items,
        upstream_tasks=[exists_items],
    )

#tasks.py

@task(max_retries=30, retry_delay=timedelta(minutes=1))
def wait_list_creation(item_id: str) -> bool:
    item_status = try_create_item(item_id)
    if item_status:
        return item_id
    raise Exception(f"item_id didn't create")

@task(trigger=any_successful)
def used_upstream_data(input_data, exists_items) -> dict:
    res = do_something(input_data, exists_items)
    return res

print_mapped_output

from datetime import timedelta
from prefect import Flow, task
from prefect.executors import LocalDaskExecutor
from prefect.triggers import any_successful
from prefect.tasks.notifications import SlackTask
from random import random


@task(log_stdout=True, trigger=any_successful)
def print_mapped_output(x):
    message = f"Final result: {x}"
    print(message)
    SlackTask(message=message).run()
    return x


@task(
    log_stdout=True, max_retries=2, retry_delay=timedelta(seconds=2),
)
def randomly_fail(x):
    print(f"Got input: {x}")
    nr = random()
    print(f"Number deciding whether task fails or not: {nr}")
    if nr > 0.5:
        raise ValueError("Too big number!")
    return x * 10


with Flow(
    name="flow_with_tasks_failing_sometimes_running_final_task_once",
    executor=LocalDaskExecutor(),
) as flow:
    numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9]
    results = randomly_fail.map(numbers)
    print_mapped_output(results)