try ... catch

from prefect import task, flow, allow_failure


@task
def fail():
    raise ValueError()


@task
def succeed(x):
    return x


@flow
def example():
    try:
        fail()
    except ValueError:
        print("Ah it failed")

    print("All good!")

    state = fail(return_state=True)
    if state.is_failed():
        print("Ah it failed")

    print("All good")

    future = fail.submit()
    try:
        future.result()
    except ValueError:
        print("Ah it failed")

    print("All good!")

    return succeed(allow_failure(future))


example()

shell_run_command