Sylvain Hazard
10/19/2021, 2:44 PMfrom 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 ?Kevin Kho
10/19/2021, 2:54 PMSylvain Hazard
10/19/2021, 2:56 PMget_list
task as an input of the get_list_results
task ? (Sorry, I'm very bad at naming tasks 😅Kevin Kho
10/19/2021, 2:56 PMSylvain Hazard
10/19/2021, 3:00 PM