Skip to content

Complex Deps

This example shows how to create more complex dependencies between tasks.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from hera.workflows import DAG, Workflow, script


@script()
def foo(p):
    if p < 0.5:
        raise Exception(p)
    print(42)


with Workflow(generate_name="complex-deps-", entrypoint="d") as w:
    with DAG(name="d"):
        A = foo(name="a", arguments={"p": 0.6})
        B = foo(name="b", arguments={"p": 0.3})
        C = foo(name="c", arguments={"p": 0.7})
        D = foo(name="d", arguments={"p": 0.9})

        A >> [B, C]
        # here, D depends on A, B, and C. If A succeeds and one of B or C fails, D still runs
        [A, (B | C)] >> D

 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
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: complex-deps-
spec:
  entrypoint: d
  templates:
  - name: d
    dag:
      tasks:
      - name: a
        template: foo
        arguments:
          parameters:
          - name: p
            value: '0.6'
      - name: b
        depends: a
        template: foo
        arguments:
          parameters:
          - name: p
            value: '0.3'
      - name: c
        depends: a
        template: foo
        arguments:
          parameters:
          - name: p
            value: '0.7'
      - name: d
        depends: a && (b || c)
        template: foo
        arguments:
          parameters:
          - name: p
            value: '0.9'
  - name: foo
    inputs:
      parameters:
      - name: p
    script:
      image: python:3.10
      source: |-
        import os
        import sys
        sys.path.append(os.getcwd())
        import json
        try: p = json.loads(r'''{{inputs.parameters.p}}''')
        except: p = r'''{{inputs.parameters.p}}'''

        if p < 0.5:
            raise Exception(p)
        print(42)
      command:
      - python

Comments