Instantiating Tasks inside flow

[DanSava]

I understand from the documentation you are not supposed to instantiate Tasks inside a flow without using them but in the example:

class AddTask(Task):
    def __init__(self, default: int, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.default = default

    def run(self, x: int, y: int=None) -> int:
        if y is None:
            y = self.default
        return x + y

with Flow("My flow!") as flow:
    add_1 = AddTask(default=1)
    add_2 = AddTask(default=2)
    first_result = add_1(x=1, y=2)
    second_result = add_2(x=first_result, y=100)

state = flow.run()

In my opinion, both instances of the AddTask class are called inside the flow but I still get the following warning:

UserWarning: Tasks were created but not added to the flow: {<Task: Task>}. This can occur when `Task` classes, including `Parameters`, are instantiated inside a `with flow:` block but not added to the flow either explicitly or as the input to another task. For more information, see https://docs.prefect.io/core/advanced_tutorials/task-guide.html#adding-tasks-to-flows.

Is this the intended behaviour?

[zangell44]

Hi @DanSava thanks for taking the time to write this up!

Unfortunately, I'm not sure there's an easy way to fix this warning. I'll do a little more digging and open a GitHub issue if a solution makes sense.

Generally, we recommend initializing tasks outside the flow for readability and clarity on initializing vs. calling tasks. This warning was added to help users that get stuck and is safe to ignore.

[zanieb]

It turns out this was a really sneaky bug! Fix at #4368

[DanSava]

Thank you for taking the time to replay and looking into it!