Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Screen Shot 2022-02-23 at 4.54.33 PM.png

is there a way i can avoid the superfluous `List` and `Dict` tasks in my DAG? this is taken from a barely modified example on prefect’s blog (code in thread)

```from typing import List

from prefect import Flow, Parameter, task
from prefect.engine.results import LocalResult
from prefect.tasks.prefect import create_flow_run, get_task_run_result
from prefect.executors import LocalExecutor


@task(result=LocalResult(), log_stdout=True)
def create_some_data(length: int):
    print(length)
    return list(range(length))

@task(result=LocalResult(), log_stdout=True)
def minimum(data: List[int]) -&gt; int:
    new = data[0] + 2
    print("new", new)
    return new

with Flow("child", executor=LocalExecutor()) as child_flow:
    data_size = Parameter("data_size", default=5)
    data = create_some_data(data_size, task_args=dict(slug="foobar"))
    minimum(data, task_args=dict(slug="min"))

@task(log_stdout=True)
def transform_and_show(data: List[int]) -&gt; List[int]:
    print(f"Got: {data!r}")
    new_data = [x + 1 for x in data]
    print(f"Created: {new_data!r}")
    return new_data

with Flow("parent", executor=LocalExecutor()) as parent_flow:
    child_run_id = create_flow_run(
        flow_name=child_flow.name, parameters=dict(data_size=10)
    )
    child_min = get_task_run_result(child_run_id, "min")
    new_child_run_id = create_flow_run(
        flow_name=child_flow.name, parameters=dict(data_size=child_min)
    )
    child_data = get_task_run_result(new_child_run_id, "foobar")

    transform_and_show(child_data)```

check out <https://discourse.prefect.io/t/how-to-avoid-tasks-such-as-list-tuple-or-dict-in-a-flow-structure/318|this page> - you can basically create an intermediary task to avoid those extra tasks such as List or Dict

thanks <@U02H1A95XDW>! i’m still a little confused why the `List` exists at all, but it looks like the following change cleans everything up :slightly_smiling_face:
```@task
def get_flow_parameters(data_size):
    return dict(data_size=data_size)

with Flow("parent", executor=LocalExecutor()) as parent_flow:
    child_run_id = create_flow_run(
        flow_name=child_flow.name, parameters=get_flow_parameters(data_size=10)
    )
    child_min = get_task_run_result(child_run_id, "min")
    new_child_run_id = create_flow_run(
        flow_name=child_flow.name, parameters=get_flow_parameters(data_size=child_min)
    )
    child_data = get_task_run_result(new_child_run_id, "foobar")

    transform_and_show(child_data)```

it looks like this is because `child_min` isn’t an `int` as i was expecting but rather a `&lt;class 'prefect.tasks.core.function.FunctionTask'&gt;` instead

True, the result of the task, when you pass it to downstream tasks, will be int but until then this is a FunctionTask, you're correct.