How to retrieve results/output from workflow tasks after the workflow is complete

[rgasper]

If I create and submit a workflow, how can I pull results out from the tasks? In a related note, is there a way to define at the workflow level what the "Outputs" are, instead of the client having to know which specific tasks to pull which specific output from for later use?

e.g.

# my_workflow.py
from hera.workflows import Workflow, script


@script()
def hello(s: str) -> str:
    return("Hello, {s}!".format(s=s))


with Workflow(
    generate_name="hello-world-",
    entrypoint="hello",
    arguments={"s": "world"},
) as w:
    hello()

# submit_script.py
from my_workflow import w

w.arguments = {"s": "Hera"}
w.create(wait=True)
if not (w.status and w.status.phase == "Succeeded"):
     raise RuntimeError('workflow failed')

# what next?

output_from_w =filter( 
    lambda n: n.display_name == "hello",
    m.status.nodes.values(),  # why is the list of nodes in the status attribute? copied from https://hera.readthedocs.io/en/stable/examples/workflows/use-cases/testing_templates_and_workflows/
)[0].???? # how do I access the output from the task

What I'm trying to achieve is having a single python function that takes in some arguments, submits them into a workflow for execution on different containers, waits for the workflow to finish, then returns the final result. Do all outputs have to be artifacts? is the way to access those outputs different if they are not artifacts?

[elliotgunton]

To answer

is there a way to define at the workflow level what the "Outputs" are

first - I don't think there is. A Workflow is just an abstraction to hold templates, which themselves have inputs and outputs.

how can I pull results out from the tasks?

For parameters, you can take inspiration from this test:

hera/tests/submissions/test_optional_input_parameter.py

Lines 50 to 51 in 0d283df

	node: NodeStatus = next(filter(lambda n: n.display_name == step, model_workflow.status.nodes.values()))
	message_out: ModelParameter = next(filter(lambda n: n.name == "message-out", node.outputs.parameters))

The node.outputs follows the spec for Outputs https://argo-workflows.readthedocs.io/en/stable/fields/#outputs, so you can find Parameters (name and value) and artifacts (name only, the value is on your artifact storage).

For Artifacts, you can then use a WorkflowsService to get_output_artifact:

hera/src/hera/workflows/service.py

Line 1355 in 0d283df

def get_output_artifact(self, name: str, node_id: str, artifact_name: str, namespace: Optional[str] = None) -> str:

[rgasper]

Follow up question - the response from get_output_artifact is what I'm guessing is a base64 encoded byte string, but I don't see any documentation of that in the function signature, docstring, or the argo-workflows API docs. Could you point me in the right direction for how to decode the object?

e.g this is a response I got from the apo. It has real data, but nothing sensitive, just some custom python dataclass objects with junk data used for testing. What do I do with it to turn it back into a python object? I used jsonpickle.dumps to serialize the artifact.

'b\'\\x1f\\x8b\\x08\\... redacted cus why not

If I use jsonpickle.loads(...) with that as input, it just returns it, which make sense AFAIK for what that's assuming the data will look like. So clearly there's another step. Any hints?

[rgasper]

I was able to figure it out. It's the bytes of a tar.gz of the output from jsonpickle.dumps(original_object), so I was able to unzip, untar, read & encode into utf-8, and then use jsonpickle.loads(...) to recover the object.

[elliotgunton]

Nice one! Thanks for sharing 🚀