Change state of a task from failed to completed in order for flow to end as completed

Help
1

Hello,

Problem: I am creating a flow where the task needs to enter state Completed even if it raises an error. The final state of the flow if this task fails should be Completed (and not Failed). I need to raise and error in order to stop all upstream functions from executing.
Here is a dummy code, which I need help with.

import prefect


def handle_disambig_error(task_future: prefect.futures.PrefectFuture):
    task_future.wait()  # block until completion
    if task_future.get_state().is_failed():  # future.result()
        task_future._final_state = prefect.states.COMPLETED
    return task_future


@prefect.task
def fail_successfully():
    raise


@prefect.flow()
def main_flow():
    future_obj = fail_successfully().submit()
    handle_disambig_error(future_obj)


if __name__ == "__main__":
    main_flow()

Again, I cannot remove the raise from the task and return the 0 as then I would have to change multiple upstream functions. I do not want the task to end in failed state as this causes my final flow to fail. I want the flow to fail when other, more crucial tasks fail and not the mentioned one.

Thank you for your help!

2

Hello!

You can create an additional helper function that manually modifies the state of your flows based on any of the conditions you mentioned. Please refer to the docs on a code example of how to set that. Once your conditions are met, you can kick of a series of flows via a deployment

1 Like