https://prefect.io logo
Title
s

Stéphan Taljaard

08/25/2022, 1:52 PM
Hi With Prefect 1 - Any tips for detecting if a task was used within another task without using
task._*run*()_
? I want to add a pytest to check and prevent adding incorrect task uses like this.
import prefect

@prefect.task
def task1(par):
    task2()  # this is incorrect! it should be task2.run()

@prefect.task
def task2():
    return 1

with prefect.Flow("Sample flow") as flow:
    aa = prefect.Parameter("aa")
    task1(aa)
flow.validate()
doesn't return errors/warnings, since it doesn't check what's happening inside each task. Meanwhile, in the task, there's another task hiding and being used incorrectly. It will fail when running the flow.
ValueError: Could not infer an active Flow context while creating edge to <Task: task2>. This often means you called a task outside a `with Flow(...)` block. If you're trying to run this task outside of a Flow context, you need to call `task2.run(...)`
I could do a
flow.run()
in my test, but that can be cumbersome and require a lot of scaffolding in terms of setting up mocks, and even with mocks, the flow run might just be too slow. A
task.validate()
can be handly, but I'm finding some difficulty in writing such a function.