Matthew Blau01/28/2021, 2:42 PM
and I am struggling to find relevant information on how best to debug this. I thank you all in advance for helping!
Unexpected error: ValueError('Could not infer an active Flow context.')
Greg Roche01/28/2021, 3:29 PM
with Flow() as flow:
With the imperative API you have to specifically add the task to a flow, something like this:
with Flow("example") as flow: result = my_task() flow.run()
flow = Flow("example") results = my_task() flow.set_dependencies(results) flow.run()
Matthew Blau01/28/2021, 3:41 PM
with 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 Roche01/28/2021, 3:45 PM
Matthew Blau01/28/2021, 3:45 PM
Greg Roche01/28/2021, 3:56 PM
, there's probably some logic in there which is leading to a task which hasn't been added to a flow. For example if
is itself trying to call a function which has been decorated with
(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.
So you would either need to remove the
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.
, which resolves the error. Or, you would restructure a bit and have the
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 Blau01/28/2021, 4:13 PM
Greg Roche01/28/2021, 4:13 PM