Hera Python SDK

DAGs (Directed Acyclic Graphs) are composed of Tasks. Unlike Steps, which use context managers for parallelism, DAGs require you to explicitly specify dependencies between tasks using the right-shift (>>) operator. This operator tells Argo which tasks must complete before a subsequent task begins. Tasks without defined dependencies will start immediately upon Workflow execution.

Key differences from Steps:

• Explicit Dependencies: You must use >> to link tasks.
• Flexibility: You can define tasks anywhere and incrementally build the graph by adding dependencies across multiple lines or even importing tasks from other files.

from hera.workflows import DAG, Workflow, script

@script()
def echo(message):
    print(message)

with Workflow(generate_name="dag-diamond-", entrypoint="diamond") as w:
    with DAG(name="diamond"):
        A = echo(name="A", arguments={"message": "A"})
        B = echo(name="B", arguments={"message": "B"})
        C = echo(name="C", arguments={"message": "C"})
        D = echo(name="D", arguments={"message": "D"})

        A >> [B, C] >> D

Hera decorators are members of the Workflow (or WorkflowTemplate) class. They are used to link Python functions to specific Argo Workflow template types. Instead of using context managers, you declare a WorkflowTemplate object and use its methods as decorators to define templates.

Key Concepts:

• Template Declaration: You must declare a WorkflowTemplate or Workflow instance upfront (e.g., w = WorkflowTemplate(name="my-template")).
• Input/Output Handling: To declare inputs and outputs for your templates, you must use the Input and Output Pydantic classes from hera.workflows. This allows Hera to deduce the template schema from the function signature.
• Local Execution: The dag and steps decorators allow you to run and test functions locally in pure Python. However, Argo-specific features like expressions or with_items loops will not function during local execution.

from hera.workflows import Input, Output, WorkflowTemplate

w = WorkflowTemplate(name="my-template")

class MyInput(Input):
    user: str

class MyOutput(Output):
    my_str: str

@w.script()
def hello_world(my_input: MyInput) -> MyOutput:
    output = MyOutput()
    output.my_str = f"Hello {my_input.user}"
    return output

The RunnerScriptConstructor uses the Hera Runner to execute your function. This allows you to write Python code in a standard way, solving the limitations of inline scripts.

Advantages

• Standard Imports: Imports can be placed anywhere in your package.
• Modular Code: The decorated function can call other functions defined in your package.
• Advanced Types: Supports any serializable class, including Pydantic models, for both inputs and outputs.
• Clean YAML: Instead of dumping large code blocks into the Workflow YAML, it uses the hera.workflows.runner module to call your function via a module path (module:function).

Requirements

• You must build a custom Docker image containing your source code, its dependencies, and hera itself.
• The image must be accessible to your Argo cluster before submitting the Workflow.

from hera.workflows import RunnerScriptConstructor
from pydantic.v1 import BaseModel

class Input(BaseModel):
    a: int
    b: str = "foo"

class Output(BaseModel):
    output: List[Input]

@script(constructor=RunnerScriptConstructor(), image="my-code-image:v1")
def my_function(input: Input) -> Output:
    return Output(output=[input])

with Workflow(
    generate_name="hello-world-",
    entrypoint="my_function",
    arguments={"input": Input(a=42)},
) as w:
    my_function()

The Steps template is used to run a sequence of templates.

• Sequential execution: By default, templates within a Steps context run one after another.
• Parallel execution: You can run templates in parallel by using the .parallel() context manager within a Steps block.
• Invocating: You can invoke templates using Step() or by calling a @script decorated function directly within the Steps context.

from hera.workflows import Container, Parameter, Step, Steps, Workflow

with Workflow(
    generate_name="steps-",
    entrypoint="hello-hello-hello",
) as w:
    whalesay = Container(
        name="whalesay",
        inputs=[Parameter(name="message")],
        image="docker/whalesay",
        command=["cowsay"],
        args=["{{inputs.parameters.message}}"],
    )

    with Steps(name="hello-hello-hello") as s:
        Step(
            name="hello1",
            template="whalesay",
            arguments={"message": "hello1"},
        )

        with s.parallel():
            Step(
                name="hello2a",
                template="whalesay",
                arguments={"message": "hello2a"},
            )
            Step(
                name="hello2b",
                template="whalesay",
                arguments={"message": "hello2b"},
            )