Hey Everyone, Im currently trying to get a better grasp on how to best interact with past results of flows. Something similar to what metaflow has https://docs.metaflow.org/metaflow/tagging would be great. Is there some preferred way to do this? Something like:

with Flow("A Flow") as my_flow:
    task_res = my_task()
task.run()

followed by:

latest_flow_state = my_flow.last_successful_run
latest_flow_state.tasks[task_res]

Hi @Moritz Rocholl - if I understand correctly, you should be able to do this:

with Flow("A Flow") as my_flow:
    task_res = my_task()

last_flow_state = task.run()
print(last_flow_state)
# <Success: "All reference tasks succeeded.">

the return of

flow.run

will contain the state of the run and all its associated tasks 🙂

Hey. thanks for the reply. You are right that is the obvious solution, especially if I have caching in place and can simply use the results of the last run. Nevertheless, if I am not mistaken, that only works for local runs triggered directly when called from within the same process locally. It does not work for server triggered runs, or does it?

That's correct, to access runs created through the API (Prefect Server or Cloud), you'll need to make a query to the API, something like:

query {
  flow_run(where: { flow_id: { _eq: "<<YOUR FLOW ID>>" } } ) {
    id
    state
    # any other data you're interested in
  }
}

Thank you! Il look into that. On another note, do you know of any resource on how to best structure result storage? Does prefect have a stance regarding that?

Hm, I think we typically structure results according to flow run id, which ensures they're unique; otherwise the structure is largely up to how you need to parse the data later.