https://prefect.io logo
j

Jacqueline Riley Garrahan

02/18/2022, 6:52 PM
Hi ๐Ÿ‘‹, I am using
prefect.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?
k

Kevin Kho

02/18/2022, 7:01 PM
Hi @Jacqueline Riley Garrahan, can you show how are you using it? Are you using it inside a Flow?
j

Jacqueline Riley Garrahan

02/18/2022, 7:02 PM
Copy code
with Flow("schedule-run") as flow:
        flow_run_id = create_flow_run(
                        flow_name=flow_name,
                        project_name=project_name, 
                        parameters=data
                    )
Yup!
k

Kevin Kho

02/18/2022, 7:04 PM
This looks right. How are you trying to use the
flow_run_id
?
j

Jacqueline Riley Garrahan

02/18/2022, 7:07 PM
I'd like to poll for completion of the task:
Copy code
prefect.tasks.prefect.flow_run.get_task_run_result(flow_run_id)
k

Kevin Kho

02/18/2022, 7:19 PM
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?
j

Jacqueline Riley Garrahan

02/18/2022, 7:45 PM
@Kevin Kho exactly what i wanted ๐Ÿ™
Thank you!
๐Ÿ‘ 1
Also, this project is awesome
k

Kevin Kho

02/18/2022, 7:45 PM
Thank you!
j

Jacqueline Riley Garrahan

02/18/2022, 7:46 PM
@Kevin Kho if using wait_for_flow_run, can I get the result from the run?
k

Kevin Kho

02/18/2022, 7:48 PM
You have to use the
get_task_run_result
but that is a specific task inside the subflow. You should find an example here
j

Jacqueline Riley Garrahan

02/18/2022, 8:05 PM
Do I need to define a slug for tasks?
Copy code
ValueError: 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.
k

Kevin Kho

02/18/2022, 8:08 PM
There is a default slug created but you can do. The
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 UI
j

Jacqueline Riley Garrahan

02/18/2022, 8:12 PM
Could you give example? Trying to access like this doesn't work as the slug is None:
Copy code
with 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)
k

Kevin Kho

02/18/2022, 8:17 PM
Ah actually I think if you do
print(flow.serialize())
on the subflow side, you can see the task-slug it was registered with in there
Or this to be cleaner:
print(flow.serialize()['tasks'])
And look for the
slug
entries to get the slug
3 Views