I am struggling a little with Prefect programming patterns. Take this example from this doc: https://docs.prefect.io/core/advanced_tutorials/task-looping.html

with Flow("mapped-fibonacci") as mapped_flow:
    ms = Parameter("ms")
    fib_nums = compute_large_fibonacci.map(ms)

flow_state = mapped_flow.run(ms=[10, 100, 1000, 1500])

nice but what about non array parameters? Let's say I have two parameters, one that's an array, and the other which is a constant.

with Flow("mapped-fibonacci") as mapped_flow:
    ms = Parameter("ms")
    ms2 = Parameter("ms2")
    fib_nums = compute_large_fibonacci.map(ms, ms2)

flow_state = mapped_flow.run(ms=[10, 100, 1000, 1500])

You can use unmapped for that

Awesome. Two more questions: • Does the functional API support expressing task dependencies without explicitly adding edges to the flow? • How would you code this: run ten tasks, and when completed, ten more tasks

For the first . For the second, you have to split into two different tasks and then use

upstream_tasks=[unmapped(task_one)]

to force all the task_one to complete before task_two

Nice for question 1. Is upstream_tasks a kwarg args that can be used with functions annotated with @task ? The second question, seems like task composition is a daunting topic when learning prefect

It is used during the Flow so it’s not attached to the

@task

. For the second, are the first ten tasks and second tasks different?

yes

It has to be in the flow block:

with Flow("example") as flow:
    a = first_task()
    b = second_task()
    c = third_task(c_inputs, upstream_tasks=[a,b])

Would that work if third task is defined like that?

@task def third_task(c): 
         pass

Yes exactly. Upstream/downstream are always defined in the Flow, not at the task level

@Kevin Kho can parameters come from env variables?

You can use

EnvVarSecret

instead?

make sense, thank you

A parameter is a special task that is injected at run time. Parameters don’t read from env variables, but you can based on the default:

@task
def get_val(x):
    if x == "test:
        return os.environ["something"]
    else:
        return "something_else"
with Flow(...) as flow:
    x = Parameter("x", "test")
    get_val(x)