Jacqueline Riley Garrahan
02/18/2022, 6:52 PMprefect.tasks.prefect.create_flow_run
to kick off some tasks. I've noticed that this isn't returning an id for the flow run as documented and instead returning a Task
object. Any advice on how to access the ids of created runs?Kevin Kho
Jacqueline Riley Garrahan
02/18/2022, 7:02 PMwith Flow("schedule-run") as flow:
flow_run_id = create_flow_run(
flow_name=flow_name,
project_name=project_name,
parameters=data
)
Kevin Kho
flow_run_id
?Jacqueline Riley Garrahan
02/18/2022, 7:07 PMprefect.tasks.prefect.flow_run.get_task_run_result(flow_run_id)
Kevin Kho
get_task_run_result
is meant to be on a task level inside the flow_run_id
. Do you want wait_for_flow_run
instead?Jacqueline Riley Garrahan
02/18/2022, 7:45 PMKevin Kho
Jacqueline Riley Garrahan
02/18/2022, 7:46 PMKevin Kho
get_task_run_result
but that is a specific task inside the subflow. You should find an example hereJacqueline Riley Garrahan
02/18/2022, 8:05 PMValueError: No task runs found while querying for task runs where {'task': {'slug': {'_eq': 'create_some_data-1'}}, 'flow_run_id': {'_eq': '601f6e54-b2fb-4119-b9c3-1faaf0a00205'}, 'map_index': {'_eq': -1}}
[2022-02-18 12:02:16-0800] INFO - prefect.TaskRunner | Task 'get_task_run_result': Finished task run for task with final state: 'Failed'
[2022-02-18 12:02:16-0800] INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.
Kevin Kho
slugifying
is kind of aggressive so it happens stuff like -1 or -copy sometimes. The API is the best way to get the slug. Iโm not sure if itโs in the UIJacqueline Riley Garrahan
02/18/2022, 8:12 PMwith Flow("schedule-run") as flow:
flow_run_id = create_flow_run(
flow_name=flow_name,
project_name=project_name,
parameters=data
)
# need slug
child_data = get_task_run_result(flow_run_id, flow_run_id.slug)
Kevin Kho
print(flow.serialize())
on the subflow side, you can see the task-slug it was registered with in thereprint(flow.serialize()['tasks'])
slug
entries to get the slug