KG
07/01/2024, 4:56 PMMarvin
07/01/2024, 4:56 PMpython
from prefect import flow, task
@task
def my_task():
raise ValueError("An error occurred!")
@flow
def my_subflow():
my_task()
@flow
def my_parent_flow():
try:
my_subflow()
except Exception as e:
print(f"Subflow failed with error: {e}")
# Handle failure or continue execution
if __name__ == "__main__":
my_parent_flow()
This ensures that the parent flow catches the exception and handles it without failing entirely.
For more details, you can check Prefect documentation or useful examples at this link.Nate
07/01/2024, 5:09 PMreturn_state=True
to trap failed states like
from prefect import flow, task
@task
def bad():
raise ValueError("asdf")
@flow
def ok_with_failure():
state = bad(return_state=True)
assert state.is_failed()
return "its fine"
ok_with_failure()
...
12:08:54.113 | INFO | Flow run 'fiery-jackal' - Finished in state Completed()
Out[3]: 'its fine'