Add Default configuration for submitting Tasks to the Defined TaskRunner

[masonmenges]

Describe the current behavior

Currently when executing a prefect flow tasks are not executed by the task runner unless you explicitly use the .submit or .map methods on a task.

The following example for instance does not submit the task to the task runner on task a but does on task b since it's explicitly called

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def some_task(input: str):
    print("Task: " + input)
    print("task doing task things")


@flow(task_runner=ThreadPoolTaskRunner) 
def some_flow():
    some_task("a")
    some_task.submit("b").wait()

some_flow()

Describe the proposed behavior

In practice being able to explicitly state when you do and don't want something to be submitted to the task runner makes perfect sense but there are scenarios where the expectation is that every task should be run against the task runner, being able to define some form of global default wether via an env variable or an arg to the flow that assumes each task should be submitted to the task runner when it executes.

Example Use

No response

Additional context

In a general context the concern here is mostly around consistency of behavior, needing to call an additional method on every task opens up the possibility for incorrect execution behavior when the intent is for every task to execute on the task runner.

[zzstoatzz]

hi @masonmenges - is the suggestion that there'd be a setting where if set, would give the following behavior?

@flow(task_runner=ThreadPoolTaskRunner) 
def some_flow():
    some_task("a") # submits to task runner
    some_task.submit("b").wait() # also submits to task runner

if so, IMO I'm not sure that's something we'd want to add, as its pretty inexplicit, in a way we've intentionally tried to move away from in prefect 3.

[rh-km]

Plus one to this, I agree that the default state makes sense, but there should at least be some kind of option or way to make it happen. I messed around with a custom decorator but did not have much luck. Something like this might work but I suspect has some serious downsides:

def background_task(task_func: Callable) -> Callable:

    def as_background_task(*args, **kwargs):
        return task(task_func).submit()

    return as_background_task

[zzstoatzz]

Thanks for sharing your perspective @rh-km! I understand the desire to reduce boilerplate when you want tasks to consistently run on the task runner. However, I'd like to explain our reasoning for maintaining the explicit .submit() pattern:

The explicit submission model was an intentional design choice in Prefect 3 to make task execution behavior clear and predictable at the call site. While .submit() adds a few extra characters, it serves as clear documentation of intent/behavior directly in the code.

If you're looking to reduce boilerplate, a helper function in your own code could be a good middle ground. However, you'd want to be careful about the implementation details to preserve the full task functionality. Here's a potential pattern:

def submit_task(task_fn, *args, **kwargs):
    return task_fn.submit(*args, **kwargs)

@flow
def my_flow():
    # More explicit than auto-submission, less verbose than .submit()
    result = submit_task(some_task, "arg")

This maintains explicit control over task submission while reducing the repetitive .submit() calls that may have motivated this request

---

note that the issue as stated above doesn't consider whether futures would be automatically resolved (i.e. .result()), which is where the conversation gets more complicated, because if you did auto-resolve futures, then you'd effectively have my_task.submit().result() and the actual behavior would be no different than the current behavior of calling my_task(). If you didn't, you would have another dual / contextual scenario where sometimes you have to call .result() on the return value of a called task, and sometimes you wouldn't (this dual / context behavior has been something we've been meaning to intentionally remove in prefect 3)

let me know if that is helpful context

[aaazzam]

I'm going to convert this to a discussion for us to better understand where to land on do-it-yourself vs. magic. We've purposefully swung the pendulum, hard, away from magic to do-it-yourself -- since magic makes things hard to introspect, but there's likely room for us to reach a sensible middle ground.

But! For now this isn't planned.

[cicdw]

Yea I'm curious what the use case for this is -- if we did this, there would be no execution benefit, because the submitted work would need to also be resolved immediately, preventing any form of concurrency or parallelism.