@Marvin I am trying to use https://docs.prefect.io/latest/concepts/states/#state-change-hooks to perform an action on specific states such as failures, but in the example https://docs.prefect.io/latest/guides/state-change-hooks/#example-use-cases it says that flows will retry before the failure hooks are triggered, is that also the case with tasks? What I want is a hook that performs an action (such as waiting on another event) at each retry that a failure occurs, not after all retries have failed. How can I implement this with a hook.

In Prefect, state change hooks can indeed be used to perform actions when task states change, including retries due to task failures. However, information regarding the behavior of state change hooks for task retries specifically is not directly mentioned in the excerpts provided. To implement a hook that performs an action at each retry after a failure, you can attach a state change hook to the task itself. Prefect allows you to define hooks for various state transitions. The hook for the retry state can be set to trigger an action whenever the task attempts a retry due to a failure. Since we lack direct examples from the documentation regarding this behavior, it may be best to create a custom state change hook on your task that listens for retry attempts. In Prefect, this is done using the

@task

decorator's

state_handlers

parameter, where you can define a function that handles the transition. Below is a basic example of how you might define such a function:

python
from prefect import task, Task, flow
from prefect.engine.state import Retrying, Failed

@task
def your_task():
    # Your task code that might fail and retry
    pass

def on_retry_handler(task: Task, old_state, new_state):
    if isinstance(old_state, Failed) and isinstance(new_state, Retrying):
        # Perform your action here, like waiting on another event
        pass
    return new_state

with flow(name="your_flow") as flow:
    task_with_hook = your_task(state_handlers=[on_retry_handler])

In this example,

on_retry_handler

is a function that checks if the task is transitioning from a

Failed

to a

Retrying

state. If so, it performs the desired action. To get more detailed instructions and examples, it's recommended to check the Prefect documentation on State Change Hooks and to examine code examples from Prefect's repositories for more context on how the hooks function in practice. If official examples don't cover this precise use-case, experimentation or reaching out to the Prefect Community for further guidance might be necessary. Please note that as an AI, I'm employing my inherent knowledge and the data retrieved here rather than personal experience with Prefect 2.x, as it is beyond my training cutoff. I would recommend verifying any implementation details directly in the Prefect documentation or source code.

@Nate Do you know how I could implement this? I'm looking at objects.State in the python sdk. Does it just have to be manual in my code or can it still be done with a state hook?

do you need the logic to happen in a hook function outside the flow run, or do you just want to do something different in your task in the case of a retry?

Basically, I can have a spotty internet connection and I'm connecting to databases nonlocally just to download data, but the downloads once they start might last for a couple weeks. So before a retry I want the script to wait on something that polls my internet until it's reconnected, then reconnects the asyncpg connections I handed to the flow. I see how I can add a specific hook to a subflow with a partial which is nice and let's me avoid global stuff.

there's also this

retry_condition_fn

that might help you here

Oh that is perfect, thanks.