from prefect import flow, task
import random


@task
def on_failure_task():
    print("Taking action on failure here")


@task
def on_success_task():
    print("Taking action on success here")


@flow
def bad_subflow():
    if random.random() > 0.5:
        raise ValueError("Non-deterministic error has occured.")


@flow(log_prints=True, name="Conditional subflows")
def main():
    state = bad_subflow(return_state=True)
    if state.name == "Failed":
        print("Subflow failed! Run on_failure_task")
        on_failure_task()
    else:
        print("Subflow succeeded! Run on_success_task")
        on_success_task()


if __name__ == "__main__":
    main()