Matthew Blau
01/28/2021, 2:42 PMUnexpected error: ValueError('Could not infer an active Flow context.')
and I am struggling to find relevant information on how best to debug this. I thank you all in advance for helping!Greg Roche
01/28/2021, 3:29 PMwith Flow() as flow:
context, e.g.
with Flow("example") as flow:
result = my_task()
flow.run()
With the imperative API you have to specifically add the task to a flow, something like this:
flow = Flow("example")
results = my_task()
flow.set_dependencies(results)
flow.run()
Matthew Blau
01/28/2021, 3:41 PMwith Flow("example",
storage = Docker(dockerfile="/home/mblau/projects/integration/Dockerfile", ignore_healthchecks= True
)) as flow:
main()
flow.run_config = DockerRun(image = "integration_mosaic_integration:latest")
flow.register(project_name="test")
Greg Roche
01/28/2021, 3:45 PMmain()
have a @task()
decorator?Matthew Blau
01/28/2021, 3:45 PMGreg Roche
01/28/2021, 3:56 PMmain()
, there's probably some logic in there which is leading to a task which hasn't been added to a flow. For example if main()
is itself trying to call a function which has been decorated with @task()
(I think this is considered an anti-pattern and instead the called task should be added explicitly to the flow).ValueError: Could not infer an active Flow context.
from prefect import task, Flow
@task()
def foo():
print("foo")
@task()
def main():
foo()
with Flow("example") as flow:
main()
flow.run() # ValueError: Could not infer an active Flow context.
So you would either need to remove the @task()
decorator from foo()
, which resolves the error. Or, you would restructure a bit and have the foo()
task explicitly added to the flow, e.g.
@task()
def foo():
return "foo"
@task()
def main(arg):
print(arg)
with Flow("example") as flow:
what_to_print = foo()
main(what_to_print)
flow.run() # "foo"
Matthew Blau
01/28/2021, 4:13 PMGreg Roche
01/28/2021, 4:13 PM