<@ULVA73B9P> If I have a flow with a loop to execu...
# ask-marvin
g
@Marvin If I have a flow with a loop to execute many subflows. How to retry failed subflows and continue remaining subflows? The default behavior is stop the flow run whenever any subflow fails.
m
Handling retries of failed subflows while continuing with remaining subflows in Prefect 3.x requires careful orchestration of exceptions and task states. Prefect 3.x allows more flexibility in handling failures without causing the entire flow to fail. Here are some strategies you can apply: 1. Retry Failed Subflows: You can configure tasks or subflows to retry on failure. Prefect allows you to specify retry logic directly in your tasks or subflows, such as the number of retries and a delay between retries. 2. Exception Handling: To prevent a failure in a subflow from stopping the entire flow, use exception handling to manage expected failures and define a fallback or alternate flow of execution. 3. Continue on Failure: By default, a flow will not fail unless an exception propagates. This means that subflows can fail, but as long as their exceptions are handled or suppressed (e.g., using
raise_on_failure=False
), the main flow will continue executing remaining subflows. 4. ControlFlow and Custom Termination Conditions: ControlFlow, an agentic framework built on top of Prefect, allows for more complex control logic, including custom termination conditions and the use of the
run_until
parameter for greater flexibility. For deeper insights, check out the Prefect 3.x documentation on flow orchestration and retry behavior and explore the ControlFlow patterns for running tasks. Remember, it’s always good to specify how resilient each part of your pipeline needs to be and how errors should be mitigated to ensure smooth flow execution.
n
hi @Gary - you can use
return_state=True
to return the state object that you can use to decide to raise the exception or not
Copy code
In [1]: from prefect import flow

In [2]: @flow
   ...: def f():
   ...:     raise ValueError("I failed!")
   ...:

In [3]: state = f(return_state=True)
21:06:23.143 | INFO    | prefect.engine - Created flow run 'laughing-python' for flow 'f'
....
    raise ValueError("I failed!")
ValueError: I failed!
21:06:23.208 | ERROR   | Flow run 'laughing-python' - Finished in state Failed('Flow run encountered an exception: ValueError: I failed!')

In [4]: if state.is_completed():
   ...:     print('not gonna happen')
   ...: elif state.is_failed():
   ...:     print('its ok')
   ...:
its ok
g
It works, thank you Nate!
n
catjam