@Marvin what is a way i can emulate the usage of custom_state_handler from prefect 1 in prefect 2

In Prefect 2, you can build custom logic that reacts to task-run states within your flow function without the need for state_handlers. You can raise an arbitrary exception in your task or flow and return a custom state. For more details and examples, you can refer to the Prefect documentation on "How to take action on a state change of a task run" and "How can I stop the task run based on a custom logic".

there are state hooks in prefect 2, like

def some_hook(flow, flow_run, state):
   # do whatever
   pass

@flow(on_completion=[some_hook])

gotcha, thanks, i’m trying to set up a task that does some work that isn’t core to the main purpose of the flow but if that sate bombs out for some reason don’t actually stop the flow

that could work, or on flow or task calls you can use

return_state=True

to suppress exceptions raised by tasks or flows so they dont raise in the calling parent

@task
def foo():
   raise ValueError()

@flow(log_prints=True)
def resilient():
   state = foo(return_state=True)

   if state.is_failed():
      print("its ok")
      return 42

Gotcha. Thanks!

if you go the run deployment route ill just mention that

run_deployment

blocks until the underlying flow finishes by default, unless you pass

timeout=0

(fire and forget)

thanks for the heads up, yeah i just want to shoot this off into the void. Probably eventually it’ll shoot over to some kinda job queue like sqs