Hi, i'm trying to get the output from an upstream task (task_a). How do i so in this example? I want the rand_num to show when it is successful. How can I retrieve it through a flow?

#!/usr/bin/env python

import random

from prefect.triggers import all_successful, all_failed
from prefect import task, Flow

@task(name="Task A")
def task_a():
    rand_num = float(random.random())
    curr_limits = float(0.5)
    if rand_num < curr_limits:
        raise ValueError(f'{rand_num} is less than {curr_limits}')
    return rand_num

@task(name="Task B", trigger=all_successful)
def task_b():
    pass

@task(name="Task C", trigger=all_failed)
def task_c():
    pass

if __name__ == '__main__':
    with Flow('My triggers') as flow:
        success = task_b(upstream_tasks=[task_a])
        fail = task_c(upstream_tasks=[task_a])

    flow.set_reference_tasks([success])
    flow_state = flow.run()

Do something like this where task_b takes in the value and outputs it:

@task(name="Task B", trigger=all_successful)
def task_b(value):
    print(value)


with Flow('My triggers') as flow:
    rand_num = task_a()
    success = task_b(rand_num)
    fail = task_c(rand_num)

etc...

Thanks Josh, I've actually went that route while waiting for a reply. What if you have multiple upstream tasks and you need to obtain the output of one of those tasks? How would you code this?

Multiple upstream tasks all feeding into a single downstream task?

Got it . So all the logic is done within the context manager, and that includes the passing of parameters.

In this case, yes, but Flows don’t have to be defined purely in this sense and you may use any mix of both the functional and imperative API!

oh ...