https://prefect.io logo
Title
i

itay livni

12/02/2019, 8:23 PM
Hi - Is there a way to get the result of a specific merge task after running a flow . Similar to this:
flow_state.result[ifelse_return].result
n

Nancy

12/02/2019, 8:27 PM
Hi @itay livni we’ve got an example here, does this help? https://docs.prefect.io/api/unreleased/tasks/control_flow.html#functions
i

itay livni

12/02/2019, 8:30 PM
merge
?
n

Nancy

12/02/2019, 8:34 PM
with Flow("My Flow"):
        true_branch = ActionIfTrue()
        false_branch = ActionIfFalse()
        ifelse(CheckCondition(), true_branch, false_branch)

        merged_result = merge(true_branch, false_branch)
i

itay livni

12/02/2019, 8:37 PM
I need to rephrase the question. How do I access merged result after running "My Flow"
merged_result
is in another file.
get_defs_state.result[merge].result
j

josh

12/02/2019, 8:42 PM
Is that not returning your result? The access pattern from above works for me:
In [10]: flow_state.result[merged_result].result
Out[10]: True
Not sure what you mean by
merged_result
being in another file
i

itay livni

12/02/2019, 8:45 PM
Yes that works when running a flow from the same file the flow is defined in. However if the flow is run from a different file\module then an error (obviously occurs) `Using variable before assignment
I tried get tasks but the merged results was not defined in the flow
j

josh

12/02/2019, 8:49 PM
If you are calling the flow and trying to retrieve the result from a different process then it will need to be persisted somewhere via a result handler https://docs.prefect.io/core/concepts/results.html
The merged result will also only exist on a flow object (if you’re trying to access it that way) once it is run. i.e. importing the flow will not contain the merge result
i

itay livni

12/02/2019, 8:52 PM
# Get Disambiguated Definitions
get_defs_state = get_defs_flow.run(
        parameters=dict(
            term=args.term,node_artifact_id=node_artifact_id,
            node_artifact_flpth=node_artifact_flpth
        )
    )
    get_defs_flow.visualize(flow_state=get_defs_state)

#definitions_df = get_defs_flow.get_tasks()
    definitions_df = get_defs_state.result[definitions_df].result
j

josh

12/02/2019, 8:54 PM
Ah okay I see what you’re trying to do. Now I understand
i

itay livni

12/02/2019, 8:56 PM
ok cool. So what is the right way to do this
the result handler section does not have any code examples 😞
j

josh

12/02/2019, 9:00 PM
Try:
task = get_defs_flow.get_tasks()[...] # get the task
get_defs_state.result[task].result
Where you get the task you want from the list returned by
get_tasks
i

itay livni

12/02/2019, 9:01 PM
so that task is not in the list. I tried that
neither is merge
j

josh

12/02/2019, 9:02 PM
Hmm weird because it appears in the flow for me:
In [21]: flow.get_tasks()
Out[21]:
[<Task: CompareValue: "True">,
 <Task: check>,
 <Task: Merge>,
 <Task: True>,
 <Task: CompareValue: "False">,
 <Task: False>]
If you visualize your flow is the merge task present?
i

itay livni

12/02/2019, 9:08 PM
yes... That worked 🙂 Not sure why it was failing earlier.
thank you
j

josh

12/02/2019, 9:09 PM
Great! I opened an issue to add some documentation on the result handler page because even though it was unrelated to this we definitely should add some clarity there 🙂
i

itay livni

12/02/2019, 9:09 PM
yes I looked at if before trying this method to see if it was appropriate and well ...
Thanks again