#4573 [Core feature] Add support try and except in the workflow Issue created by pingsutw Motivation: Why do you think this is important? It's follow-up to these failure node PRs • flyteorg/flytekit#840 • #4308 People might want to pass a specific task output to a failure node. To support that flytekit need to support try and except first, so users are be able to pass a promise to failure node. Goal: What should the final outcome look like, ideally?

@workflow
def wf(name: str = "my_cluster"):
    id = create_cluster(name=name)
    try:
        t = t1(a=1, b="2")
    except:
        d = delete_cluster(id=id)  # failure node
    c >> t >> d

Describe alternatives you've considered For now, failure node's input is same as workflow inputs.

@task(container_image=image_spec)
def clean_up(name: str):
    print(f"Deleting cluster {name}")

@workflow(on_failure=clean_up)
def wf2(name: str = "kevin"):
    c = create_cluster(name=name)
    t = t1(a=1, b="2")
    d = delete_cluster(name=name)
    c >> t >> d

Propose: Link/Inline OR Additional context No response Are you sure this issue hasn't been raised already? ☑︎ Yes Have you read the Code of Conduct? ☑︎ Yes flyteorg/flyte