Kelly Huang
02/17/2022, 1:37 AMKevin Kho
Kelly Huang
02/17/2022, 1:43 AMflow 1 = StartFlowRun(
flow_name= flow 1,
project_name=project_name, wait=True)
flow 2 = StartFlowRun(
flow_name=Flow 2,
project_name=project_name, wait=True)
flow 3 = StartFlowRun(
flow_name=Flow 3,
project_name=project_name, wait=True)
with Flow('flow '4) as flow:
PARAMS HERE
first flow = flow 1(
upstream_tasks=[recipient_email, funds_list,
company_sector, market_cap_lower_boundary,
market_cap_upper_boundary],
parameters={PARAMS})
second flow = flow 2(
upstream_tasks=[first flow],
parameters={PARAMS})
third flow = flow 3(
upstream_tasks=[second flow],
parameters={PARAMS})
Kevin Kho
Kelly Huang
02/17/2022, 1:46 AMKevin Kho
StartFlowRun
, doesn’t mean you can’t add more tasks to the “main” flow right?Kelly Huang
02/17/2022, 1:49 AMKevin Kho
with Flow('flow '4) as flow:
PARAMS HERE
first flow = flow 1(
upstream_tasks=[recipient_email, funds_list,
company_sector, market_cap_lower_boundary,
market_cap_upper_boundary],
parameters={PARAMS})
x = some_other_task_here()
Kelly Huang
02/17/2022, 4:02 AMupstream_tasks
, but the second flow complains that the task was called without its parameter, but I'm not trying to run the task again, I just want its resultKevin Kho
get_task_run_result
task to pull the task result from another Flow. DocsKelly Huang
02/22/2022, 8:22 PMflow_run_id
refer to the version group id
? and how do I find the task_slug
?get_task_run_result
still apply?Kevin Kho
create_flow_run
so it’s given by the main Flow. The second is retrieve it with get_task_run_result
.
The easiest why to get a task slug is by printing flow.serialize()['tasks']
and that should contain the slugs of the task then choose the one you want. It’s honestly hard to predict the output of the slugify
algorithm something so printing is more straightforwardwith Flow() as flow:
x = create_flow_run()
y = get_task_run_result(x, task_slug)
Or you can just do it from the main flow:
with Flow() as flow:
some_output = some_task()
x = create_flow_run(..., parameters={"out": some_output})
and then the child flow will get it
EDIT: These are not mutually exclusive. you can have bothKelly Huang
02/22/2022, 8:52 PMfrom prefect.tasks.prefect.flow_run import
I tried but it doesn't existKevin Kho
from prefect.tasks.prefect import get_task_run_result
. It’s the same spot as create_flow_run
Kelly Huang
02/22/2022, 9:11 PMcreate_flow_run
. I use from prefect.tasks.prefect import StartFlowRun