get_task_run_result with a Failed task

[Pofilo]

Hello,

I have a question about get_task_run_result() (doc here - version prefect==1.2.2).

In the doc:

You will need to ensure results can be accessed.

But how to deal with a Failed task ?

Here is task_B:

@task(result=LocalResult(location="{flow_run_id}_{task_name}_{task_run_id}.pickle"))
def task_B(param_1, param_2):
    logger = prefect_context.get("logger")
    logger.info(f"param_1={param_1} param_2={param_2}")
    if param_1 == "not-wanted-value":
        raise FAIL(message="...")
    return param_1, param_2

Here is a simple example of my flow:

with Flow("my_global_flow") as my_global_flow:
    flow_1 = create_flow_run(
        flow_name="my_flow_1",
        project_name="foo",
        parameters={"boo": "bar"},
    )
    wait_for_flow_1 = wait_for_flow_run(
        flow_1,
        stream_states=True,
        stream_logs=True,
        raise_final_state=True,
    )
    flow_1_result_1 = get_task_run_result(
        task_slug="task_B-1", flow_run_id=flow_1
    )

    flow_2 = create_flow_run(
        flow_name="my_flow_2",
        project_name="foo",
        parameters={"foo2": "bar2", "flow_1_result_1": flow_1_result_1},
    )
    wait_for_flow_2 = wait_for_flow_run(
        flow_2,
        stream_states=True,
        stream_logs=True,
        raise_final_state=True,
    )

Here, is task_B is failing, get_task_run_result() will not be able to get its result.
However, instead of just failing the flow, I'm getting this traceback:

Task 'get_task_run_result': Exception encountered during task execution!
Traceback (most recent call last):
  File "/foo/var/venv/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/foo/var/venv/lib/python3.8/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/foo/var/venv/lib/python3.8/site-packages/prefect/tasks/prefect/flow_run.py", line 239, in get_task_run_result
    return task_run.get_result()
  File "/foo/var/venv/lib/python3.8/site-packages/prefect/backend/task_run.py", line 75, in get_result
    self._assert_result_type_is_okay()
  File "/foo/var/venv/lib/python3.8/site-packages/prefect/backend/task_run.py", line 162, in _assert_result_type_is_okay
    raise ValueError(
ValueError: The task result has no `location` so the result cannot be loaded. This often means that your task result has not been configured or has been configured incorrectly.

Is it a desired behavior ? It can happens than a key task cannot succeed, this traceback and its message is not really accurate.

In get_result() from task_run.py, we check that the task is_finished, but should we also check is the task is not failed ?

[AlexisMunera98]

@Pofilo Did u solve it?

[Pofilo]

Yes, I overloaded get_task_run_result in another task checking the flow state.

@task(trigger=all_finished)
def get_flow_result(
    flow_run_view: "backend.FlowRunView", flow_run_id: str, task_slug: str
) -> Any:
    """
    Task that run no matter what thanks to the "all_finished" trigger.
    To be used instead of "get_task_run_result()" to make sure that the next
    flow/task is not run even if the flow failed.

    The problem comes from "get_task_run_result()" task which is run even if the
    previous flow failed and seems to reset the global state (further flows and tasks
    will be run because "get_task_run_result()" is in Success state).

    Instead of 2 parameters about the flow, one could use
    `FlowRunView.from_flow_run_id(flow_run_id)` but it does not work.

    :param flow_run_view: A view of the flow run after completion
    :param flow_run_id: The flow run the task run belongs to
    :param task_slug: The 'slug' of the task run you want to get the result of
    :return The return value of the task is the flow is successful, raise FAIL otherwise
    (so the rest of the flow has a TriggerFailed state).
    """
    if not isinstance(flow_run_view.state, Success):
        raise FAIL(message="The flow has failed, let's fail too.")

    return get_task_run_result.run(flow_run_id=flow_run_id, task_slug=task_slug)

Then, in my previous example, you can just replace:

    flow_1_result_1 = get_task_run_result(
        task_slug="task_B-1", flow_run_id=flow_1
    )

by

    flow_1_result_1 = get_flow_result(
        flow_run_view=wait_for_flow_1,
        flow_run_id=flow_1,
        task_slug="task_B-1"
    )

This way, the flow will be in a failed state.
Don't know if there is a better way, but the trick works.