Hi all! Does anyone have a best practice for a “prefect-equivalent” list comprehension on the result of a mapped task? Trying:

from prefect import Flow, task

@task()
def some_task(x):
    return x+1

some_list = [1, 2, 3, 4]

with Flow("list_comp") as list_comp_flow:
    mapped_result = some_task.map(some_list)
    list_comp = [result+1 for result in mapped_result]

results in the following 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.

Which makes sense of course, since while building the flow,

mapped_result

is still a

Task

, not a

list

.

Wouldnt this be

map(some_task, some_list)

instead of the other way around? EDIT: Apologies, I see prefect tasks implement a map function, Ill be following this too. Their docs here https://docs.prefect.io/core/concepts/mapping.html#prefect-approach seem to imply that you should have a function which does the result+1 so that it can be executed as a task at runtime

Thanks Samuel! That is indeed a proper solution, and in line with the Prefect mindset, and perhaps the best way to go about it. I find myself reluctant to write a task for each “small” list comp, but that might just be my own habits, and I need to get used to this 🙂

I was with you in the same mindset, having come from airflow (where waiting for a task to be picked up could take minutes and thus lots of small tasks slowed things down a ton). With a dask executor running my taks though, its been super quick, this flow of mine with dozens and dozens of tasks executres in a couple of seconds:

Here is my solution, for what it’s worth:

from prefect import Flow, task
@task()
def some_task(x):
    return x+1

@task
def list_comprehension(original_list, lambda_func):
    return [lambda_func(x) for x in original_list]

some_list = [1, 2, 3, 4]

with Flow("list_comp") as list_comp_flow:
    mapped_result = some_task.map(some_list)
    list_comp = list_comprehension(mapped_result, lambda x: x+2)

You can also just pass the lambda inside a

task()

call since the decorator is just syntactic sugar for the same thing

from prefect import Flow, task
@task()
def some_task(x):
    return x+1

some_list = [1, 2, 3, 4]
with Flow("list_comp") as list_comp_flow:
    mapped_result = some_task.map(some_list)
    list_comp = task(lambda x: x+1).map(mapped_result)

Not necessarily worse then my generic “list_comprehension” in the logs…

Note the two will have different flow graphs. The one you had makes

list_comprehension

one task while this makes another

map

operation.