@Marvin is there a way to call a flow and set the task and flow retries = 0? For testing, or manual triggers I would like a flow to fail if any task or subflow fails.

To call a flow and set the task and flow retries to 0, you need to specify the retry policy in your flow and task definitions. The retry number is defined via the

max_retries

parameter for tasks and potentially via the configuration for flows. Unfortunately, none of my searches in the Prefect documentation returned a specific example of disabling retries for a flow and task within a flow run for testing purposes. To ensure a flow fails if any task or subflow fails without retries, you can set

max_retries

to 0 when defining your tasks and ensure no retry policy is set on the flow level. Here's an example of how you can define a task with no retries:

python
from prefect import task, Flow

@task(max_retries=0)
def my_task():
    # your task implementation

with Flow("my_flow") as flow:
    my_task()

# To run the flow without any retries
flow.run()

It's important to note that the

max_retries

parameter defaults to 0 if not specified, which means the task will not retry on failure by default. Make sure to consult the Prefect documentation on tasks to confirm how retries are configured and check for any updates or additional configurations related to flow-level settings and failures. If there are concerns about specific flow runners or execution environments, consult the relevant sections to ensure the zero-retry policy will be honored as expected. If you're using Prefect Cloud, you might also need to check for settings that could affect retries within the cloud infrastructure. Please note that the retrieved information from the documentation was not directly related to setting retries to 0. It's possible that this detail might be explained in a different part of the documentation or that it's a guardrail that requires specific inquiry into the configuration options for Prefect 2.x.

Thanks well I guess I can define flows to pass retries parameters down as kwargs.

if i understand the ask, you can also use

with_options

for this

from prefect import flow

@flow
def foo():
  raise RuntimeError()

foo.with_options(retries=1)()

or I guess you wanted to set it to 0, but should be able to pick any value

whoa thats great, exactly what i was looking for! I suppose there is no way to do this with subflows or tasks within a flow?

there is indeed

from prefect import flow, task

@task
def foo():
  raise RuntimeError()

@flow
def bar(n: int):
  foo.with_options(retries=n)() # retry n times (n=3)

@flow
def baz():
  n = 3
  bar.with_options(retries=1, flow_run_name="retrying {n} times flow")(n) # retry once

baz()