Hello all, I have been successful in getting prefe...
# prefect-community
m
Hello all, I have been successful in getting prefect to build and run a flow on a docker container. However, I am getting an error of
Copy code
Unexpected 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!
g
This happens when you're trying to execute a task, but the task hasn't been assigned to a flow, so Prefect doesn't know how to run the task in relation to other tasks. The most relevant documentation is probably https://docs.prefect.io/core/concepts/flows.html#apis but in short, if you're using the functional API, you should specify your task within a
with Flow() as flow:
context, e.g.
Copy code
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:
Copy code
flow = Flow("example")
results = my_task()
flow.set_dependencies(results) 
flow.run()
m
@Greg Roche so for reference my flow code shows
Copy code
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")
g
does
main()
have a
@task()
decorator?
m
yes
g
Then at this point I would start looking into what's going on in
main()
, 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).
For example, this snippet also gives me
ValueError: Could not infer an active Flow context.
Copy code
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.
Copy code
@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"
m
@Greg Roche awesome, that was the issue. I am fairly new to prefect and am still learning all the concepts but that works! I have other issues to resolve but that's more with the code of the file than prefect at this point, I think
g
Glad I could help 🙂