Thread
#prefect-community
    Jacqueline Riley Garrahan

    Jacqueline Riley Garrahan

    7 months ago
    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?
    Kevin Kho

    Kevin Kho

    7 months ago
    Hi @Jacqueline Riley Garrahan, can you show how are you using it? Are you using it inside a Flow?
    Jacqueline Riley Garrahan

    Jacqueline Riley Garrahan

    7 months ago
    with Flow("schedule-run") as flow:
            flow_run_id = create_flow_run(
                            flow_name=flow_name,
                            project_name=project_name, 
                            parameters=data
                        )
    Yup!
    Kevin Kho

    Kevin Kho

    7 months ago
    This looks right. How are you trying to use the
    flow_run_id
    ?
    Jacqueline Riley Garrahan

    Jacqueline Riley Garrahan

    7 months ago
    I'd like to poll for completion of the task:
    prefect.tasks.prefect.flow_run.get_task_run_result(flow_run_id)
    Kevin Kho

    Kevin Kho

    7 months ago
    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

    Jacqueline Riley Garrahan

    7 months ago
    @Kevin Kho exactly what i wanted ๐Ÿ™
    Thank you!
    Also, this project is awesome
    Kevin Kho

    Kevin Kho

    7 months ago
    Thank you!
    Jacqueline Riley Garrahan

    Jacqueline Riley Garrahan

    7 months ago
    @Kevin Kho if using wait_for_flow_run, can I get the result from the run?
    Kevin Kho

    Kevin Kho

    7 months ago
    You have to use the
    get_task_run_result
    but that is a specific task inside the subflow. You should find an example here
    Jacqueline Riley Garrahan

    Jacqueline Riley Garrahan

    7 months ago
    Do I need to define a slug for tasks?
    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.
    Kevin Kho

    Kevin Kho

    7 months ago
    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
    Jacqueline Riley Garrahan

    Jacqueline Riley Garrahan

    7 months ago
    Could you give example? Trying to access like this doesn't work as the slug is None:
    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)
    Kevin Kho

    Kevin Kho

    7 months ago
    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