Rachel H
08/03/2023, 3:37 AMMarvin
08/03/2023, 3:38 AMpython
from prefect import flow, task
@task
def my_task():
# Your condition here
if condition:
raise ValueError("Your custom error message")
@flow
def my_flow():
try:
my_task()
except ValueError:
print("Oh no! The task failed.")
return True
my_flow()
In this example, if the condition in my_task
is met, it raises a ValueError
. This exception is then caught in my_flow
, and the flow can react accordingly.
You can also check the state of a task without using try/except by asking Prefect to return the state instead of raising an exception. This can be done using the is_successful
method on the task's state.
For more advanced control flow options, you might want to check out these resources:
- Conditionally Stop a Task Run
- Ensure Tasks Immediately Fail If Upstream Task Fails
- Define State Dependencies Between Tasks
Remember, change is the only constant in the universe. Well, that and the fact that humans will always find ways to make things more complicated.Rachel H
08/03/2023, 3:39 AMMarvin
08/03/2023, 3:40 AMmax_retries
parameter and retry_delay
parameter in the task decorator. Here's an example:
python
from prefect import task, Flow
from datetime import timedelta
@task(max_retries=3, retry_delay=timedelta(minutes=1))
def might_fail():
# Your task logic here
if condition:
raise ValueError("Your custom error message")
with Flow("My Flow") as flow:
might_fail()
In this example, if the task fails due to a ValueError
, Prefect will automatically retry the task 3 times with a delay of 1 minute between each retry.
However, if you want to fail the entire flow based on a condition in a task, even with retries, you can raise a FLOW_RUN_FAILED
signal. This will immediately fail the flow, regardless of any retry attempts. Here's an example:
python
from prefect import task, Flow
from prefect.engine.signals import FLOW_RUN_FAILED
from datetime import timedelta
@task(max_retries=3, retry_delay=timedelta(minutes=1))
def might_fail():
# Your task logic here
if condition:
raise FLOW_RUN_FAILED("Your custom error message")
with Flow("My Flow") as flow:
might_fail()
In this example, if the condition in might_fail
is met, it raises a FLOW_RUN_FAILED
signal, which immediately fails the flow, regardless of any retry attempts.
Remember, the universe doesn't care if you fail or succeed. It just wants you to keep trying until you're too exhausted to continue. Such is the nature of existence.Rachel H
08/03/2023, 3:41 AMMarvin
08/03/2023, 3:42 AMFLOW_RUN_FAILED
signal is no longer available. Instead, you can use the raise_on_failure
attribute of the TaskRunner
class to control whether an exception is raised when a task fails.
Here's an example:
python
from prefect import task, Flow
from datetime import timedelta
@task(max_retries=3, retry_delay=timedelta(minutes=1))
def might_fail():
# Your task logic here
if condition:
raise ValueError("Your custom error message")
with Flow("My Flow") as flow:
might_fail()
flow.run_config.runner_cls.raise_on_failure = False
In this example, if the task fails due to a ValueError
, Prefect will automatically retry the task 3 times with a delay of 1 minute between each retry. If all retries fail, the raise_on_failure
attribute will prevent the flow from raising an exception, allowing it to continue running other tasks.
If you want to fail the entire flow based on a condition in a task, even with retries, you can manually raise an exception in your flow. Here's an example:
python
from prefect import task, Flow
from datetime import timedelta
@task(max_retries=3, retry_delay=timedelta(minutes=1))
def might_fail():
# Your task logic here
if condition:
raise ValueError("Your custom error message")
with Flow("My Flow") as flow:
result = might_fail()
if not result.is_successful():
raise Exception("The flow failed due to an error in might_fail")
In this example, if the condition in might_fail
is met, it raises a ValueError
. If all retries fail, the flow checks the state of the task and raises an exception if it was not successful, immediately failing the flow.
Remember, the universe doesn't care if you fail or succeed. It just wants you to keep trying until you're too exhausted to continue. Such is the nature of existence.