https://prefect.io logo
Join SlackCommunities
Powered by
Hi community! I’m creating my own instance of a Ta...
# ask-community
j
Hi community! I’m creating my own instance of a Task class that returns multiple values, I added the Tuple return-type annotation but I’m still getting the error `TypeError: Task is not iterable. If your task returns multiple results, pass 
nout
to the task decorator/constructor, or provide a 
Tuple
return-type annotation to your task.` when I instantiate the Task with nout=2. Class in the thread. Any suggestions?
The task attached gets instantiated as follows:
Copy code
get_product = GetProduct(nout=2, checkpoint=False)
multiple_return_task.py
k
Hey @Julio Venegas! I’ve seen this error come up also depending how the output of the task is used downstream. Does it give you an error when this is the ending task of your Flow?
j
Should’ve mentioned this: the error is raised when I try to register the flow that is using the task
k
Is your flow code too long to post?
j
It is quite a complicated flow haha
k
Ok i’ll just to recreate with a minimal example
j
here the flow part to spare you more lines
multiple_returns_flow.py
line 38 is where 
get_product
gets used in the flow @Kevin Kho
the instantiation is the one I mentioned above 
get_product = GetProduct(nout=2, checkpoint=False)
k
ok i’ll try to re-create
will take a bit
j
no worries, thank you!
k
Copy code
from typing import Tuple
from prefect import Task, Flow, task, flatten


@task
def get_collection():
    return [0,1]

@task
def get_collection_details(n):
    return {"name": f"name{n}"}

@task
def get_collection_item_list(n):
    return {"id": n, "created": "2019-01-01"}

@task
def pair_item_collection(collection_item, collection_detail):
    return (collection_item, collection_detail)

class GetProduct(Task):
    def run(self, item_collection_pair: tuple) -> Tuple[str, dict]:
        collection_item, collection_details = item_collection_pair
        if not collection_item:
            return None, {}  
        else:
            product = {'collection_item_id': collection_item['id'],
                       'collection_item_created_at': collection_item['created'],
                       }
            return collection_item['id'], {**collection_details, **product}

get_product = GetProduct(nout=2, checkpoint=False)

with Flow('test') as flow:
    collection_response = get_collection()
    collection_details = get_collection_details.map(collection_response)
    collection_item = get_collection_item_list.map(collection_response)
    item_collection_details_pairs = pair_item_collection.map(collection_item, collection_details)
    product_id, collection_product = get_product.map(flatten(item_collection_details_pairs))

flow.run()
Can you confirm this is a minimum example we can go off of?
j
Looks good @Kevin Kho!
k
Ok will work on it
j
thank you!
I’m in CEST time, so I’ll be going to sleep soon but will check here back tomorrow in case you need any more details from my side.
k
Copy code
class GetProduct(Task):
    def run(self, item_collection_pair: tuple) -> Tuple[str, dict]:
        collection_item, collection_details = item_collection_pair
        if not collection_item:
            return None, {}  
        else:
            product = {'collection_item_id': collection_item['id'],
                       'collection_item_created_at': collection_item['created']}
            return collection_item['id'], product

get_product = GetProduct(nout=2)

with Flow('test') as flow0:
    collection_response = get_collection()
    collection_details = get_collection_details.map(collection_response)
    collection_item = get_collection_item_list.map(collection_response)
    item_collection_details_pairs = pair_item_collection.map(collection_item, collection_details)
    test = get_product.map(item_collection_details_pairs)
    product_ids = task(lambda x: x[0]).map(test)
    collection_product = task(lambda x: x[1]).map(test)

flow0.run()
The explanation is that the output of a map operation is 
Iterable
so it can be 
Iterable[Tuple]
but the way you were calling it is 
Tuple[Iterable]
where you would be expecting two lists.
a, b = ([1,2,3], [2,3,4])
in Python works but 
a,b = [(1,2),(2,3),(3,4)]
will not 🙂
j
Ok, I think I get it. It then means that for all tasks that use map, if I want to return multiple values, I always need to add one task per returned value that gets the Nth item from the initial Iterable, right @Kevin Kho?
If I had not used map, then the documented behavior here: https://docs.prefect.io/core/concepts/tasks.html#multiple-return-values would have worked
k
Yes that’s right because 
map
takes in an Iterable and outputs an Iterable so you need something to reshape
j
Perfect, makes sense! Thanks a lot for taking the time to explain!
k
Happy to help!
🙌 1
2 Views
Open in Slack
PreviousNext
Powered by