Hi Team, a question about Functional API, can I specify upstream of a task manually? I want to have a task which is just wait for a few seconds, and it’s upstream of a task executes some actions

Hi @Alfie is this what you’re after?

In [4]: with Flow('dep') as flow:
   ...:     a = a()
   ...:     b = b()
   ...:     b.set_upstream(a)

(there’s also

.set_downstream()

)

thanks Josh! but my task is not a class, but a function with decorator

Yeah those are functions 🙂

In [1]: from prefect import task, Flow

In [2]: @task
   ...: def a():
   ...:     pass
   ...:

In [3]: @task
   ...: def b():
   ...:     pass
   ...:

In [4]: with Flow('dep') as flow:
   ...:     a = a()
   ...:     b = b()
   ...:     b.set_upstream(a)

great, let me try

The functional and imperative API are completely interchangeable!

got it. it works, thanks a lot!