Is there a good way to have a task generate two li...
# prefect-community
Is there a good way to have a task generate two lists (e.g. error_items, and a processed_items as output and map each list to a different downstream task?
Hi @Jacques! You can index the output of a task like this:
Copy code
from prefect import task, Flow
def two_lists():
    return {
        'list_one': [1, 2, 3],
        'list_two': [101, 102, 103]

def add_one(x):
    return x + 1

with Flow('example') as flow:
    lists = two_lists()
    x =['list_one'])
    y =['list_two'])

state =
assert state.result[x].result == [2, 3, 4]
assert state.result[y].result == [102, 103, 104]
My example returns a dictionary, but if you return a list or tuple and index as
lists[0], lists[1]
it’ll also work
this also works if you don’t want to return a
(I use something similar in my flow)
Copy code
both_items = task_that_generates_two_lists(..) # returns two lists

error_items, processed_items = both_items[0], both_items[1]
upvote 1
Awesome, thanks guys! What does this look like using the imperative api? do I just do a
error_items_task.set_upstream(two_lists['list_one'], key="data", mapped=True)
Nice, thanks!