set_flow_run_state

client.set_flow_run_state(
        flow_run_id=flow_run_id,
        state=Scheduled()
)

retries

triggers

import random
import prefect
from prefect import task, Flow
from prefect.client import Client
from prefect.tasks.notifications import SlackTask
from prefect.triggers import all_successful
from typing import cast


def post_to_slack_and_cancel_run_on_task_failure(task, old_state, new_state):
    if new_state.is_failed():
        flow_run_id = prefect.context.flow_run_id
        if isinstance(new_state.result, Exception):
            value = "```{}```".format(repr(new_state.result))
        else:
            value = cast(str, new_state.message)
        msg = (
            f"The task `{prefect.context.task_name}` failed "
            f"in a flow run {flow_run_id} "
            f"with an exception {value}. Cancelling the flow run and manually resuming later."
        )
        SlackTask(message=msg).run()
        client = Client()
        client.cancel_flow_run(flow_run_id)
    return new_state


@task(state_handlers=[post_to_slack_and_cancel_run_on_task_failure])
def run_process_that_may_fail():
    if random.random() > 0.5:
        raise ValueError("Failing due to missing information")


@task(trigger=all_successful)
def run_if_success():
    print("Success")


with Flow("state_handler_ex") as flow:
    first_task = run_process_that_may_fail()
    run_if_success(upstream_tasks=[first_task])

if __name__ == "__main__":
    flow.run()

all_successful