Whats the best way to handle graceful shutdown of a flow? I am using the parent-child deployed flow pattern but noticed that some of my flows are still running if the parent is canceled/fails. I want to catch any termination signals and ensure that those child flows also get canceled. Any advice?

Hey Matt, first thing that comes to mind is creating an automation to cancel the child flow runs in response to the parent entering a state of

Failed

or

Canceled

.

hmm unfortunately we are self-hosted and don't have access to automations currently

Hmm..you may be able to leverage something like a flow-level state handler then. Something like this:

import prefect
from prefect import task, flow
from prefect.blocks.notifications import SlackWebhook
from prefect.orion.schemas.states import StateType
import time

#Load your desired block, this example uses a slack notification block.
def notify_function():
    slack_webhook_block = SlackWebhook.load("my-test-notification")
    #This is the message you want to send to slack
    slack_webhook_block.notify("Hello from Prefect!")

#A simple flow
@flow
def my_flow():
    time.sleep(10)
    #pass
    #raise Exception

if __name__ == "__main__":
    #Store the state of your flow
    state = my_flow(return_state=True)
    '''This if statement checks for a specific flow state.
    If the condition is satisfied, the block is invoked.
    State type can be set to any available state type: <https://docs.prefect.io/concepts/states/#state-types''>'
    if state.type == (StateType.COMPLETED):
       notify_function()

We also support an

on_failure

(and

on_crashed

) state change hook that allows you to execute arbitrary Python code in response to the above terminal states. Example usage can be found here: https://github.com/PrefectHQ/prefect/blob/main/RELEASE-NOTES.md#on_completion-and-on_failure-hooks-for-flows-and-tasks