https://prefect.io logo
Title
a

Amit Singh

03/23/2020, 3:27 PM
@here I'm getting the following error when calling a task from a test case in pytest. Any clues. thanks
flow = flow or prefect.context.get("flow", None)
        if not flow:
>           raise ValueError("Could not infer an active Flow context.")
E           ValueError: Could not infer an active Flow context.
j

josh

03/23/2020, 3:28 PM
Hey @Amit Singh how are you calling the task? You may need to use the task’s
.run()
function when outside of the context of a flow
e.g.:
In [2]: @task
    ...: def a():
    ...:     print(5)
    ...:

In [3]: a()
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-17-8d7b4527e81d> in <module>
----> 1 a()

~/Desktop/code/prefect/src/prefect/core/task.py in __call__(self, mapped, task_args, upstream_tasks, flow, *args, **kwargs)
    395         new = self.copy(**(task_args or {}))
    396         new.bind(
--> 397             *args, mapped=mapped, upstream_tasks=upstream_tasks, flow=flow, **kwargs
    398         )
    399         return new

~/Desktop/code/prefect/src/prefect/core/task.py in bind(self, mapped, upstream_tasks, flow, *args, **kwargs)
    447         flow = flow or prefect.context.get("flow", None)
    448         if not flow:
--> 449             raise ValueError("Could not infer an active Flow context.")
    450
    451         self.set_dependencies(

ValueError: Could not infer an active Flow context.

In [4]: a.run()
5
c

Chris White

03/23/2020, 3:29 PM
Calling a task instance attempts to bind keyword dependencies, so calling a task instance can only occur when there is a known flow to track those dependencies; calling
task.run
like @josh mentioned avoids such internals
a

Amit Singh

03/23/2020, 3:31 PM
I have defined defined the task in a like this
@task(name='create_athena_table_from_file_task')
def create_athena_table_from_file()
c

Chris White

03/23/2020, 3:31 PM
this behavior isn’t related to your task definition
a

Amit Singh

03/23/2020, 3:31 PM
And calling it from pytest like this
response = sm_lib.athena_wrapper.create_athena_table_from_file(**temp_params)
c

Chris White

03/23/2020, 3:32 PM
see what i said above about calling a task instance
(and josh’s great example)
a

Amit Singh

03/23/2020, 3:35 PM
Got it. but i think .run is called only when the task is defined as a class, while i have created a task function here
@Chris White @josh
c

Chris White

03/23/2020, 3:36 PM
the task decorator is just a convenience for creating a Task class and instantiating it for you, so your
create_athena_table_from_file
object is actually an instance of a task class
and so it also has a run method
a

Amit Singh

03/23/2020, 3:38 PM
Thanks a lot guys. It worked
💯 2
c

Chris White

03/23/2020, 3:38 PM
awesome - anytime!