https://prefect.io logo
Title
m

Manuel Mourato

09/14/2020, 2:58 PM
Hello guys I have a scenario that fails with
Could not infer an active Flow context.
1) I have task1, which outputs a result that is a list , say : ["a","z"]
2) Now, I have a task2, that depends on the result of task1, but should only receive the first element of the result list as an input

Example:
@task
def replace_str(x):
    return x.replace("a","A")

@task
def create_str_list():
    return ["a","b","c"]
flow= Flow("test")
flow.set_dependencies(create_str_list)

--> THIS IS WHAT I WANT TO DO
flow.set_dependencies(replace_str, keyword_tasks={"x": create_str_list[0]})
Is this possible? Thank you
c

Chris White

09/14/2020, 3:22 PM
Hi Manuel - yes this is possible, with a small change to your code. Note that
create_str_list[0]
is actually creating a new task which will index into your
create_str_list
at runtime. It is this task creation that is causing the error you are seeing. To resolve it, you can enter a Flow context:
with flow:
    flow.set_dependencies(replace_str, keyword_tasks={"x": create_str_list[0]})