https://prefect.io
Join Slack
Hi :wave:, I am using `prefect.tasks.prefect.create_flow_run` to kick off some tasks. I've noticed t...
# ask-community
j
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
Hi @Jacqueline Riley Garrahan, can you show how are you using it? Are you using it inside a Flow?
j
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
This looks right. How are you trying to use the 
flow_run_id
?
j
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
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
@Kevin Kho exactly what i wanted 🙏
Thank you!
👍 1
Also, this project is awesome
k
Thank you!
j
@Kevin Kho if using wait_for_flow_run, can I get the result from the run?
k
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
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
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
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
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
2 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by