Skip to content

Template Refs

This example shows how to use TemplateRef with WorkflowTemplates and ClusterWorkflowTemplates.

Note, running this example will attempt to create the WorkflowTemplate and ClusterWorkflowTemplate on the cluster.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from hera.exceptions import AlreadyExists
from hera.workflows import (
    ClusterWorkflowTemplate,
    Step,
    Steps,
    Workflow,
    WorkflowsService,
    WorkflowTemplate,
    script,
)
from hera.workflows.models import TemplateRef


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


with WorkflowTemplate(name="hello-world-workflow-template", namespace="argo") as wt:
    hello()


with ClusterWorkflowTemplate(name="hello-world-cluster-workflow-template") as cwt:
    hello()


with Workflow(
    generate_name="use-workflow-templates-",
    entrypoint="steps",
    namespace="argo",
) as w:
    with Steps(name="steps"):
        Step(
            name="hello-template-ref",
            template_ref=TemplateRef(
                name="hello-world-workflow-template",
                template="hello",
                cluster_scope=False,
            ),
            arguments={"s": "this is using a WorkflowTemplate"},
        )
        Step(
            name="hello-cluster-template-ref",
            template_ref=TemplateRef(
                name="hello-world-cluster-workflow-template",
                template="hello",
                cluster_scope=True,
            ),
            arguments={"s": "this is using a ClusterWorkflowTemplate"},
        )

if __name__ == "__main__":
    WORKFLOWS_SERVICE = WorkflowsService(
        host="https://localhost:2746",
        verify_ssl=False,
    )
    try:
        wt.workflows_service = WORKFLOWS_SERVICE
        wt.create()
    except AlreadyExists:
        print(f"{wt.name} already exists")

    try:
        cwt.workflows_service = WORKFLOWS_SERVICE
        cwt.create()
    except AlreadyExists:
        print(f"{cwt.name} already exists")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: use-workflow-templates-
  namespace: argo
spec:
  entrypoint: steps
  templates:
  - name: steps
    steps:
    - - name: hello-template-ref
        arguments:
          parameters:
          - name: s
            value: this is using a WorkflowTemplate
        templateRef:
          name: hello-world-workflow-template
          clusterScope: false
          template: hello
    - - name: hello-cluster-template-ref
        arguments:
          parameters:
          - name: s
            value: this is using a ClusterWorkflowTemplate
        templateRef:
          name: hello-world-cluster-workflow-template
          clusterScope: true
          template: hello

Comments