@Marvin I am trying to override a flow's

on_failure

hook with

with_options

, but when I override the hook and deploy the flow with

deploy

the new

on_failure

hook is not used. Is there something else I have to do to make the new hook take effect?

To override a flow's

on_failure

hook using

with_options

, you must ensure that you're using the proper syntax when setting up your flow. The

on_failure

hook should be a list of functions that you want to be called when a failure state occurs. However, when you are deploying, you need to ensure that your deployment script or process includes this modification. Here's a snippet of how you might define this in the code:

python
@flow(on_failure=[hook1, hook2])
def my_flow():
    pass

Make sure your failure hook logic is correctly defined as a list even if there's only one function. If you are dynamically overriding the hook, ensure the

with_options

method is used correctly and is part of the flow definition before deployment. Unfortunately, I didn't find an exact match for

with_options

not working in the context of

deploy

, but these are the general guidelines. You should review the relevant documentation for more details. Check out State Change Hooks and consider looking at example test flows such as in the Prefect GitHub repository to guide your implementation. If the issue persists, consider verifying other deployment configurations and ensuring there are no overriding settings in your environment that might affect the deployment of flow hooks.

This is what I'm attempting

@flow
def some_flow():
  raise ValueError("I always fail")

def fail_hook(flow, flow_run, state):
  print("Hook ran")

async def main():
  some_flow.with_options(on_failure=[fail_hook]).deploy(
    # ...
  )

But when I run the flow from the resulting deployment, the

fail_hook

is never triggered, even though the

some_flow

flow raises an exception.