Question on testing and referring to task results. This example is contrived but maybe useful for discussion. Simplified class that defines a flow:

@task
def times_two(x):
    return x * 2

@task
def add(items):
    return sum(items)

class SimpleFlow(SRMPrefectFlow):
    @property
    def flow(self) -> Flow:
        with Flow("SimpleFlow", environment=env) as flow:
            x = Parameter("x", default=[1, 2, 3])

            times_two_task_result = times_two.map(x)
            flow_result = add(times_two_task_result)
        return flow

And a simple test to run the flow & check the last task's result:

def test_flow_run_result():
    flow = SimpleFlow().flow
    fr = flow.run()
    assert list(fr.result.values())[2].result == 12

The

list(fr.result.values())[2].result

works, but is fragile. We'd rather

fr.result[flow_result].result

but

flow_result

isn't available outside of the function that defines the flow. Is there a better approach that people have used?

That’s an interesting question @Joe Schmid, and I see where you’re coming from. There are a couple ways you could grab

flow_result

according to a convention (you may know these)

Thanks @Jeremiah

flow.get_task(name='key task name')

should work well or I like

flow.flow_result = flow_result

also. Much appreciated!

:yes: just note that since names are not guaranteed unique,

get_task(name=...)

will return a list, so as long as you choose a good name, you’ll know what the one item is (which is why this depends slightly on adopting a convention)

Makes sense. I suspect for most reasonably complex Flows we won't typically be testing them as a complete run (instead we'd unit test portions) but for simple flows this might apply. It's also nice that we have lots of example tests to refer to as part of Prefect itself.

If you come up with a good idiom here let us know 🙂

Will do!

Tags? I use them to get a particular task result of Flow when I stitch two three Flows together for organizational purposes.

res = my_special_task.get_tasks(tags=["special_tag"])[0]
df = flow_state.result[res].result
    pprint(f"Number of Definitions fron secondary sources: {df.shape}")