A WorkflowTemplate output forwarding complete example

[EricBoix]

If we conceive a WorkflowTemplate as the closest thing to a function in AW, then we might need that function to return some values. Here is a complete example (as opposed to an Hera code snipet) that

1. defines a increment-template WorkflowTemplate that forwards an output
2. calls the increment-template template and catches back it's output in the calling context.

from hera.workflows import (
    DAG,
    models as m,
    Parameter,
    script,
    Task,
    Workflow,
    WorkflowTemplate,
)
from hera.expr import g as expr


@script()
def increment(a: str):
    try:
        a_int = int(a)
        print(a_int + 1)
    except ValueError:
        print("Unable to convert ", a, " to an integer. Task failed.")
        raise Exception("failure")  # Fail task at AW level


@script()
def print_messsage(msg: str):
    print("Received: ", msg)


with WorkflowTemplate(
    name="increment-workflow-template",
    entrypoint="increment-template",
) as w:
    with DAG(
        name="increment-template",
        inputs=Parameter(name="value"),
    ) as main_dag:
        t1: Task = increment(
            name="increment-once",
            arguments=[
                Parameter(name="a", value="{{inputs.parameters.value}}")
            ],
        )
        t2 = increment(
            name="increment-twice",
            # Just to illustrate the syntax when _within_ a DAG template
            arguments=[Parameter(name="a", value=t1.result)],
        )
        t1 >> t2
        # And now the syntax to transmit a result coming from this DAG
        # to the calling context (aka "outputs forwarding").
        # First part: register the desired output within the canonical template outputs.
        expression = expr.tasks["increment-twice"].outputs.result
        main_dag.outputs = [
            Parameter(
                name="computed", value_from={"expression": str(expression)}
            )
        ]
hera_clear_workflow_template(cluster, "increment-workflow-template")
w.create()


with Workflow(
    generate_name="workflow-template-with-output-",
    entrypoint="main",
) as w:
    with DAG(name="main"):
        t1 = Task(
            name="call-workflowtemplate",
            template_ref=m.TemplateRef(
                name="increment-workflow-template",
                template="increment-template",
            ),
            arguments=Parameter(name="value", value=40),
        )

        t2 = print_messsage(
            name="print-result",
            arguments=Parameter(
                name="msg",
                # Second part: get your hands on the WorkflowTemplate output
                value="{{tasks.call-workflowtemplate.outputs.parameters.computed}}",
            ),
        )
        t1 >> t2
w.create()

Notes:

• I'm glad (and quite surprisingly proud when considering this is a shameful almost two days effort) to provide/contribute this example to the community. So feel free to do whatever usage you might feel useful/fruitful with it (as far as you consider it has any illustrative value for newbies like me: I know I'm being cocky on this one ;).
• for the record I was unable to make things working when using Steps instead of Dag.