https://prefect.io logo
Title
s

Sylvain Hazard

10/19/2021, 2:44 PM
Fell like I'm being annoying asking multiple question everyday but I can't wrap my head around this one. I've been trying to play around with the FlowRunView API and I'm running into issues with TaskRuns locations. Code and explanation below ⬇️
from prefect import task, Flow
import FilterTask
from prefect.backend import FlowRunView

import prefect

@task
def get_list():
    return [1, 2, 3, 4, 5, 6]


@task
def print_number(x):
    print(x)


@task()
def get_list_results():
    logger = prefect.context.get("logger")
    flow_run_id = prefect.context.get("flow_run_id")
    flow_run = FlowRunView.from_flow_run_id(flow_run_id)

    get_list_task = flow_run.get_task_run("get_list-1")
    <http://logger.info|logger.info>(get_list_task.get_result())


with Flow("test") as flow:
    numbers = get_list()
    prints = print_number.map(numbers)
    get_list_results(upstream_tasks=[prints])
Flow seems pretty simple but for some reason, I get a "The task result has no location" error. Shouldn't the task result have a default (local) location ? What can I do to prevent this ?
k

Kevin Kho

10/19/2021, 2:54 PM
Hey @Sylvain Hazard, no worries. I don’t think you have to do this? If they are in the same flow, you still have this data in memory right?
s

Sylvain Hazard

10/19/2021, 2:56 PM
You mean I should pass the result of the
get_list
task as an input of the
get_list_results
task ? (Sorry, I'm very bad at naming tasks 😅
k

Kevin Kho

10/19/2021, 2:56 PM
Yeah exactly. You shouldn’t have to fetch this using the API
s

Sylvain Hazard

10/19/2021, 3:00 PM
I see. I'm trying to remember why I did not go for the easy route but I believe I just fell down a rabbit hole. Thanks for getting me out !
👍 1