Skip to content

Cluster workflow template

ClusterWorkflowTemplate

ClusterWorkflowTemplates are cluster scoped templates.

Since cluster workflow templates are scoped at the cluster level, they are available globally in the cluster.

Source code in src/hera/workflows/cluster_workflow_template.py 
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
68
69
70
71
72
73
74
75
class ClusterWorkflowTemplate(WorkflowTemplate):
    """ClusterWorkflowTemplates are cluster scoped templates.

    Since cluster workflow templates are scoped at the cluster level, they are available globally in the cluster.
    """

    @validator("namespace", pre=True, always=True)
    def _set_namespace(cls, v):
        if v is not None:
            raise ValueError("namespace is not a valid field on a ClusterWorkflowTemplate")

    def create(self) -> TWorkflow:  # type: ignore
        """Creates the ClusterWorkflowTemplate on the Argo cluster."""
        assert self.workflows_service, "workflow service not initialized"
        return self.workflows_service.create_cluster_workflow_template(
            ClusterWorkflowTemplateCreateRequest(template=self.build())  # type: ignore
        )

    def get(self) -> TWorkflow:
        """Attempts to get a workflow template based on the parameters of this template e.g. name + namespace."""
        assert self.workflows_service, "workflow service not initialized"
        assert self.name, "workflow name not defined"
        return self.workflows_service.get_cluster_workflow_template(name=self.name)

    def update(self) -> TWorkflow:
        """Attempts to perform a workflow template update based on the parameters of this template.

        Note that this creates the template if it does not exist. In addition, this performs
        a get prior to updating to get the resource version to update in the first place. If you know the template
        does not exist ahead of time, it is more efficient to use `create()` directly to avoid one round trip.
        """
        assert self.workflows_service, "workflow service not initialized"
        assert self.name, "workflow name not defined"
        # we always need to do a get prior to updating to get the resource version to update in the first place
        # https://github.com/argoproj/argo-workflows/pull/5465#discussion_r597797052

        template = self.build()
        try:
            curr = self.get()
            template.metadata.resource_version = curr.metadata.resource_version
        except NotFound:
            return self.create()
        return self.workflows_service.update_cluster_workflow_template(
            self.name,
            ClusterWorkflowTemplateUpdateRequest(template=template),  # type: ignore
        )

    def lint(self) -> TWorkflow:
        """Lints the ClusterWorkflowTemplate using the Argo cluster."""
        assert self.workflows_service, "workflow service not initialized"
        return self.workflows_service.lint_cluster_workflow_template(
            ClusterWorkflowTemplateLintRequest(template=self.build())  # type: ignore
        )

    def build(self) -> TWorkflow:
        """Builds the ClusterWorkflowTemplate and its components into an Argo schema ClusterWorkflowTemplate object."""
        # Note that ClusterWorkflowTemplates are exactly the same as WorkflowTemplates except for the kind which is
        # handled in Workflow._set_kind (by __name__). When using ClusterWorkflowTemplates via templateRef, clients
        # should specify cluster_scope=True, but that is an intrinsic property of ClusterWorkflowTemplates from our
        # perspective.
        return _ModelClusterWorkflowTemplate(**super().build().dict())

active_deadline_seconds 

active_deadline_seconds: Optional[int] = None

affinity 

affinity: Optional[Affinity] = None

annotations 

annotations: Optional[Dict[str, str]] = None

api_version 

api_version: Optional[str] = None

archive_logs 

archive_logs: Optional[bool] = None

arguments 

arguments: ArgumentsT = None

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact_repository_ref 

artifact_repository_ref: Optional[ArtifactRepositoryRef] = None

automount_service_account_token 

automount_service_account_token: Optional[bool] = None

creation_timestamp 

creation_timestamp: Optional[Time] = None

deletion_grace_period_seconds 

deletion_grace_period_seconds: Optional[int] = None

deletion_timestamp 

deletion_timestamp: Optional[Time] = None

dns_config 

dns_config: Optional[PodDNSConfig] = None

dns_policy 

dns_policy: Optional[str] = None

entrypoint 

entrypoint: Optional[str] = None

executor 

executor: Optional[ExecutorConfig] = None

finalizers 

finalizers: Optional[List[str]] = None

generate_name 

generate_name: Optional[str] = None

generation 

generation: Optional[int] = None

hooks 

hooks: Optional[Dict[str, LifecycleHook]] = None

host_aliases 

host_aliases: Optional[List[HostAlias]] = None

host_network 

host_network: Optional[bool] = None

image_pull_secrets 

image_pull_secrets: ImagePullSecretsT = None

kind 

kind: Optional[str] = None

labels 

labels: Optional[Dict[str, str]] = None

managed_fields 

managed_fields: Optional[List[ManagedFieldsEntry]] = None

metrics 

metrics: MetricsT = None

name 

name: Optional[str] = None

namespace 

namespace: Optional[str] = None

node_selector 

node_selector: Optional[Dict[str, str]] = None

on_exit 

on_exit: Optional[Union[str, Templatable]] = None

owner_references 

owner_references: Optional[List[OwnerReference]] = None

parallelism 

parallelism: Optional[int] = None

pod_disruption_budget 

pod_disruption_budget: Optional[PodDisruptionBudgetSpec] = None

pod_gc 

pod_gc: Optional[PodGC] = None

pod_metadata 

pod_metadata: Optional[Metadata] = None

pod_priority 

pod_priority: Optional[int] = None

pod_priority_class_name 

pod_priority_class_name: Optional[str] = None

pod_spec_patch 

pod_spec_patch: Optional[str] = None

priority 

priority: Optional[int] = None

resource_version 

resource_version: Optional[str] = None

retry_strategy 

retry_strategy: Optional[RetryStrategy] = None

scheduler_name 

scheduler_name: Optional[str] = None

security_context 

security_context: Optional[PodSecurityContext] = None

self_link: Optional[str] = None

service_account_name 

service_account_name: Optional[str] = None

shutdown 

shutdown: Optional[str] = None

status 

status: Optional[WorkflowStatus] = None

suspend 

suspend: Optional[bool] = None

synchronization 

synchronization: Optional[Synchronization] = None

template_defaults 

template_defaults: Optional[Template] = None

templates 

templates: List[Union[Template, Templatable]] = []

tolerations 

tolerations: Optional[List[Toleration]] = None

ttl_strategy 

ttl_strategy: Optional[TTLStrategy] = None

uid 

uid: Optional[str] = None

volume_claim_gc 

volume_claim_gc: Optional[VolumeClaimGC] = None

volume_claim_templates 

volume_claim_templates: Optional[List[PersistentVolumeClaim]] = None

volumes 

volumes: VolumesT = None

workflow_metadata 

workflow_metadata: Optional[WorkflowMetadata] = None

workflow_template_ref 

workflow_template_ref: Optional[WorkflowTemplateRef] = None

workflows_service 

workflows_service: Optional[WorkflowsService] = None

Container

The Container defines a container to run on Argo.

A container generally consists of running a Docker container remotely, which is configured via fields such as image, command and args.

Container(name="cowsay", image="docker/whalesay", command=["cowsay", "foo"])

See how to use it in the Container example.

Source code in src/hera/workflows/container.py 
 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
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class Container(
    EnvIOMixin,
    ContainerMixin,
    TemplateMixin,
    ResourceMixin,
    VolumeMountMixin,
    CallableTemplateMixin,
):
    """The Container defines a container to run on Argo.

    A container generally consists of running a Docker container remotely, which is configured via fields such as
    `image`, `command` and `args`.

    ```py
    Container(name="cowsay", image="docker/whalesay", command=["cowsay", "foo"])
    ```

    See how to use it in the [Container example](../../../examples/workflows/misc/container.md).
    """

    args: Optional[List[str]] = None
    command: Optional[List[str]] = None
    lifecycle: Optional[Lifecycle] = None
    security_context: Optional[SecurityContext] = None
    working_dir: Optional[str] = None

    def _build_container(self) -> _ModelContainer:
        """Builds the generated `Container` representation."""
        assert self.image
        return _ModelContainer(
            args=self.args,
            command=self.command,
            env=self._build_env(),
            env_from=self._build_env_from(),
            image=self.image,
            image_pull_policy=self._build_image_pull_policy(),
            lifecycle=self.lifecycle,
            liveness_probe=self.liveness_probe,
            ports=self.ports,
            readiness_probe=self.readiness_probe,
            resize_policy=self.resize_policy,
            resources=self._build_resources(),
            restart_policy=self.restart_policy,
            security_context=self.security_context,
            startup_probe=self.startup_probe,
            stdin=self.stdin,
            stdin_once=self.stdin_once,
            termination_message_path=self.termination_message_path,
            termination_message_policy=self.termination_message_policy,
            tty=self.tty,
            volume_devices=self.volume_devices,
            volume_mounts=self._build_volume_mounts(),
            working_dir=self.working_dir,
        )

    def _build_template(self) -> _ModelTemplate:
        """Builds the generated `Template` representation of the container."""
        return _ModelTemplate(
            active_deadline_seconds=self.active_deadline_seconds,
            affinity=self.affinity,
            archive_location=self.archive_location,
            automount_service_account_token=self.automount_service_account_token,
            container=self._build_container(),
            daemon=self.daemon,
            executor=self.executor,
            fail_fast=self.fail_fast,
            host_aliases=self.host_aliases,
            init_containers=self._build_init_containers(),
            inputs=self._build_inputs(),
            memoize=self.memoize,
            metadata=self._build_metadata(),
            metrics=self._build_metrics(),
            name=self.name,
            node_selector=self.node_selector,
            outputs=self._build_outputs(),
            plugin=self.plugin,
            pod_spec_patch=self.pod_spec_patch,
            priority=self.priority,
            priority_class_name=self.priority_class_name,
            retry_strategy=self.retry_strategy,
            scheduler_name=self.scheduler_name,
            security_context=self.pod_security_context,
            service_account_name=self.service_account_name,
            sidecars=self._build_sidecars(),
            synchronization=self.synchronization,
            timeout=self.timeout,
            tolerations=self.tolerations,
            volumes=self._build_volumes(),
        )

active_deadline_seconds 

active_deadline_seconds: Optional[IntOrString] = None

affinity 

affinity: Optional[Affinity] = None

annotations 

annotations: Optional[Dict[str, str]] = None

archive_location 

archive_location: Optional[ArtifactLocation] = None

args 

args: Optional[List[str]] = None

automount_service_account_token 

automount_service_account_token: Optional[bool] = None

command 

command: Optional[List[str]] = None

daemon 

daemon: Optional[bool] = None

env 

env: EnvT = None

env_from 

env_from: EnvFromT = None

executor 

executor: Optional[ExecutorConfig] = None

fail_fast 

fail_fast: Optional[bool] = None

host_aliases 

host_aliases: Optional[List[HostAlias]] = None

http 

http: Optional[HTTP] = None

image 

image: Optional[str] = None

image_pull_policy 

image_pull_policy: Optional[Union[str, ImagePullPolicy]] = None

init_containers 

init_containers: Optional[List[Union[UserContainer, UserContainer]]] = None

inputs 

inputs: InputsT = None

labels 

labels: Optional[Dict[str, str]] = None

lifecycle 

lifecycle: Optional[Lifecycle] = None

liveness_probe 

liveness_probe: Optional[Probe] = None

memoize 

memoize: Optional[Memoize] = None

metrics 

metrics: Optional[MetricsT] = None

name 

name: Optional[str] = None

node_selector 

node_selector: Optional[Dict[str, str]] = None

outputs 

outputs: OutputsT = None

parallelism 

parallelism: Optional[int] = None

plugin 

plugin: Optional[Plugin] = None

pod_security_context 

pod_security_context: Optional[PodSecurityContext] = None

pod_spec_patch 

pod_spec_patch: Optional[str] = None

ports 

ports: Optional[List[ContainerPort]] = None

priority 

priority: Optional[int] = None

priority_class_name 

priority_class_name: Optional[str] = None

readiness_probe 

readiness_probe: Optional[Probe] = None

resize_policy 

resize_policy: Optional[List[ContainerResizePolicy]] = None

resources 

resources: Optional[Union[ResourceRequirements, Resources]] = None

restart_policy 

restart_policy: Optional[str] = None

retry_strategy 

retry_strategy: Optional[RetryStrategy] = None

scheduler_name 

scheduler_name: Optional[str] = None

security_context 

security_context: Optional[SecurityContext] = None

service_account_name 

service_account_name: Optional[str] = None

sidecars 

sidecars: Optional[Union[UserContainer, List[UserContainer]]] = None

startup_probe 

startup_probe: Optional[Probe] = None

stdin 

stdin: Optional[bool] = None

stdin_once 

stdin_once: Optional[bool] = None

synchronization 

synchronization: Optional[Synchronization] = None

termination_message_path 

termination_message_path: Optional[str] = None

termination_message_policy 

termination_message_policy: Optional[str] = None

timeout 

timeout: Optional[str] = None

tolerations 

tolerations: Optional[List[Toleration]] = None

tty 

tty: Optional[bool] = None

volume_devices 

volume_devices: Optional[List[VolumeDevice]] = None

volume_mounts 

volume_mounts: Optional[List[VolumeMount]] = None

volumes 

volumes: Optional[VolumesT] = None

working_dir 

working_dir: Optional[str] = None

get_artifact 

get_artifact(name: str) -> Artifact

Finds and returns the artifact with the supplied name.

Note that this method will raise an error if the artifact is not found.

Parameters:

Name Type Description Default
name str

name of the input artifact to find and return.

 required

Returns:

Name Type Description
Artifact Artifact

the artifact with the supplied name.

Raises:

Type Description
KeyError

if the artifact is not found.
Source code in src/hera/workflows/_mixins.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def get_artifact(self, name: str) -> Artifact:
    """Finds and returns the artifact with the supplied name.

    Note that this method will raise an error if the artifact is not found.

    Args:
        name: name of the input artifact to find and return.

    Returns:
        Artifact: the artifact with the supplied name.

    Raises:
        KeyError: if the artifact is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Artifact {name} not found.")
    if inputs.artifacts is None:
        raise KeyError(f"No artifacts set. Artifact {name} not found.")
    for artifact in inputs.artifacts:
        if artifact.name == name:
            return Artifact(name=name, from_=f"{{{{inputs.artifacts.{artifact.name}}}}}")
    raise KeyError(f"Artifact {name} not found.")

get_parameter 

get_parameter(name: str) -> Parameter

Finds and returns the parameter with the supplied name.

Note that this method will raise an error if the parameter is not found.

Parameters:

Name Type Description Default
name str

name of the input parameter to find and return.

 required

Returns:

Name Type Description
Parameter Parameter

the parameter with the supplied name.

Raises:

Type Description
KeyError

if the parameter is not found.
Source code in src/hera/workflows/_mixins.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def get_parameter(self, name: str) -> Parameter:
    """Finds and returns the parameter with the supplied name.

    Note that this method will raise an error if the parameter is not found.

    Args:
        name: name of the input parameter to find and return.

    Returns:
        Parameter: the parameter with the supplied name.

    Raises:
        KeyError: if the parameter is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Parameter {name} not found.")
    if inputs.parameters is None:
        raise KeyError(f"No parameters set. Parameter {name} not found.")
    for p in inputs.parameters:
        if p.name == name:
            param = Parameter.from_model(p)
            param.value = f"{{{{inputs.parameters.{param.name}}}}}"
            return param
    raise KeyError(f"Parameter {name} not found.")

DAG

A DAG template invocator is used to define Task dependencies as an acyclic graph.

DAG implements the contextmanager interface so allows usage of with, under which any Task objects instantiated will be added to the DAG’s list of Tasks.

See the DAG examples for usage.

Source code in src/hera/workflows/dag.py 
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class DAG(
    IOMixin,
    TemplateMixin,
    CallableTemplateMixin,
    ContextMixin,
):
    """A DAG template invocator is used to define Task dependencies as an acyclic graph.

    DAG implements the contextmanager interface so allows usage of `with`, under which any Task
    objects instantiated will be added to the DAG's list of Tasks.

    See the [DAG examples](../../../examples/workflows/dags/dag_diamond_with_script.md) for usage.
    """

    fail_fast: Optional[bool] = None
    target: Optional[str] = None
    tasks: List[Union[Task, DAGTask]] = []

    _node_names = PrivateAttr(default_factory=set)
    _current_task_depends: Set[str] = PrivateAttr(set())

    def _add_sub(self, node: Any):
        if not isinstance(node, Task):
            raise InvalidType(type(node))
        if node.name in self._node_names:
            raise NodeNameConflict(f"Found multiple Task nodes with name: {node.name}")
        self._node_names.add(node.name)
        self.tasks.append(node)

    def _build_template(self) -> _ModelTemplate:
        """Builds the auto-generated `Template` representation of the `DAG`."""
        tasks = []
        for task in self.tasks:
            if isinstance(task, Task):
                tasks.append(task._build_dag_task())
            else:
                tasks.append(task)
        return _ModelTemplate(
            active_deadline_seconds=self.active_deadline_seconds,
            affinity=self.affinity,
            archive_location=self.archive_location,
            automount_service_account_token=self.automount_service_account_token,
            daemon=self.daemon,
            dag=_ModelDAGTemplate(fail_fast=self.fail_fast, target=self.target, tasks=tasks),
            executor=self.executor,
            fail_fast=self.fail_fast,
            host_aliases=self.host_aliases,
            init_containers=self._build_init_containers(),
            inputs=self._build_inputs(),
            memoize=self.memoize,
            metadata=self._build_metadata(),
            metrics=self._build_metrics(),
            name=self.name,
            node_selector=self.node_selector,
            outputs=self._build_outputs(),
            parallelism=self.parallelism,
            plugin=self.plugin,
            pod_spec_patch=self.pod_spec_patch,
            priority=self.priority,
            priority_class_name=self.priority_class_name,
            retry_strategy=self.retry_strategy,
            scheduler_name=self.scheduler_name,
            security_context=self.pod_security_context,
            service_account_name=self.service_account_name,
            sidecars=self._build_sidecars(),
            synchronization=self.synchronization,
            timeout=self.timeout,
            tolerations=self.tolerations,
        )

active_deadline_seconds 

active_deadline_seconds: Optional[IntOrString] = None

affinity 

affinity: Optional[Affinity] = None

annotations 

annotations: Optional[Dict[str, str]] = None

archive_location 

archive_location: Optional[ArtifactLocation] = None

automount_service_account_token 

automount_service_account_token: Optional[bool] = None

daemon 

daemon: Optional[bool] = None

executor 

executor: Optional[ExecutorConfig] = None

fail_fast 

fail_fast: Optional[bool] = None

host_aliases 

host_aliases: Optional[List[HostAlias]] = None

http 

http: Optional[HTTP] = None

init_containers 

init_containers: Optional[List[Union[UserContainer, UserContainer]]] = None

inputs 

inputs: InputsT = None

labels 

labels: Optional[Dict[str, str]] = None

memoize 

memoize: Optional[Memoize] = None

metrics 

metrics: Optional[MetricsT] = None

name 

name: Optional[str] = None

node_selector 

node_selector: Optional[Dict[str, str]] = None

outputs 

outputs: OutputsT = None

parallelism 

parallelism: Optional[int] = None

plugin 

plugin: Optional[Plugin] = None

pod_security_context 

pod_security_context: Optional[PodSecurityContext] = None

pod_spec_patch 

pod_spec_patch: Optional[str] = None

priority 

priority: Optional[int] = None

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[RetryStrategy] = None

scheduler_name 

scheduler_name: Optional[str] = None

service_account_name 

service_account_name: Optional[str] = None

sidecars 

sidecars: Optional[Union[UserContainer, List[UserContainer]]] = None

synchronization 

synchronization: Optional[Synchronization] = None

target 

target: Optional[str] = None

tasks 

tasks: List[Union[Task, DAGTask]] = []

timeout 

timeout: Optional[str] = None

tolerations 

tolerations: Optional[List[Toleration]] = None

get_artifact 

get_artifact(name: str) -> Artifact

Finds and returns the artifact with the supplied name.

Note that this method will raise an error if the artifact is not found.

Parameters:

Name Type Description Default
name str

name of the input artifact to find and return.

 required

Returns:

Name Type Description
Artifact Artifact

the artifact with the supplied name.

Raises:

Type Description
KeyError

if the artifact is not found.
Source code in src/hera/workflows/_mixins.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def get_artifact(self, name: str) -> Artifact:
    """Finds and returns the artifact with the supplied name.

    Note that this method will raise an error if the artifact is not found.

    Args:
        name: name of the input artifact to find and return.

    Returns:
        Artifact: the artifact with the supplied name.

    Raises:
        KeyError: if the artifact is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Artifact {name} not found.")
    if inputs.artifacts is None:
        raise KeyError(f"No artifacts set. Artifact {name} not found.")
    for artifact in inputs.artifacts:
        if artifact.name == name:
            return Artifact(name=name, from_=f"{{{{inputs.artifacts.{artifact.name}}}}}")
    raise KeyError(f"Artifact {name} not found.")

get_parameter 

get_parameter(name: str) -> Parameter

Finds and returns the parameter with the supplied name.

Note that this method will raise an error if the parameter is not found.

Parameters:

Name Type Description Default
name str

name of the input parameter to find and return.

 required

Returns:

Name Type Description
Parameter Parameter

the parameter with the supplied name.

Raises:

Type Description
KeyError

if the parameter is not found.
Source code in src/hera/workflows/_mixins.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def get_parameter(self, name: str) -> Parameter:
    """Finds and returns the parameter with the supplied name.

    Note that this method will raise an error if the parameter is not found.

    Args:
        name: name of the input parameter to find and return.

    Returns:
        Parameter: the parameter with the supplied name.

    Raises:
        KeyError: if the parameter is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Parameter {name} not found.")
    if inputs.parameters is None:
        raise KeyError(f"No parameters set. Parameter {name} not found.")
    for p in inputs.parameters:
        if p.name == name:
            param = Parameter.from_model(p)
            param.value = f"{{{{inputs.parameters.{param.name}}}}}"
            return param
    raise KeyError(f"Parameter {name} not found.")

ModelMapper

Source code in src/hera/workflows/_meta_mixins.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class ModelMapper:
    def __init__(self, model_path: str, hera_builder: Optional[Callable] = None):
        self.model_path = []
        self.builder = hera_builder

        if not model_path:
            # Allows overriding parent attribute annotations to remove the mapping
            return

        self.model_path = model_path.split(".")
        curr_class: Type[BaseModel] = self._get_model_class()
        for key in self.model_path:
            fields = get_fields(curr_class)
            if key not in fields:
                raise ValueError(f"Model key '{key}' does not exist in class {curr_class}")
            curr_class = fields[key].outer_type_  # type: ignore

    @classmethod
    def _get_model_class(cls) -> Type[BaseModel]:
        raise NotImplementedError

    @classmethod
    def build_model(
        cls, hera_class: Type[ModelMapperMixin], hera_obj: ModelMapperMixin, model: TWorkflow
    ) -> TWorkflow:
        assert isinstance(hera_obj, ModelMapperMixin)

        for attr, annotation in hera_class._get_all_annotations().items():
            if mappers := get_annotated_metadata(annotation, ModelMapperMixin.ModelMapper):
                if len(mappers) != 1:
                    raise ValueError("Expected only one ModelMapper")

                # Value comes from builder function if it exists on hera_obj, otherwise directly from the attr
                value = (
                    getattr(hera_obj, mappers[0].builder.__name__)()
                    if mappers[0].builder is not None
                    else getattr(hera_obj, attr)
                )
                if value is not None:
                    _set_model_attr(model, mappers[0].model_path, value)

        return model

builder 

builder = hera_builder

model_path 

model_path = split('.')

build_model 

build_model(hera_class: Type[ModelMapperMixin], hera_obj: ModelMapperMixin, model: TWorkflow) -> TWorkflow
Source code in src/hera/workflows/_meta_mixins.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
@classmethod
def build_model(
    cls, hera_class: Type[ModelMapperMixin], hera_obj: ModelMapperMixin, model: TWorkflow
) -> TWorkflow:
    assert isinstance(hera_obj, ModelMapperMixin)

    for attr, annotation in hera_class._get_all_annotations().items():
        if mappers := get_annotated_metadata(annotation, ModelMapperMixin.ModelMapper):
            if len(mappers) != 1:
                raise ValueError("Expected only one ModelMapper")

            # Value comes from builder function if it exists on hera_obj, otherwise directly from the attr
            value = (
                getattr(hera_obj, mappers[0].builder.__name__)()
                if mappers[0].builder is not None
                else getattr(hera_obj, attr)
            )
            if value is not None:
                _set_model_attr(model, mappers[0].model_path, value)

    return model

Script

A Script in Argo Workflows acts as a wrapper around a Container, where you can specify Python code to run through source.

In Hera, you should aim to use the script decorator, rather than the Script class directly. You will need to refer to the Script class for the kwargs that the decorator can take, but your IDE should give you code completion and type hints.

Source code in src/hera/workflows/script.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
class Script(
    EnvIOMixin,
    CallableTemplateMixin,
    ContainerMixin,
    TemplateMixin,
    ResourceMixin,
    VolumeMountMixin,
):
    """A Script in Argo Workflows acts as a wrapper around a Container, where you can specify Python code to run through `source`.

    In Hera, you should aim to use the script decorator, rather than the Script class directly.
    You will need to refer to the Script class for the kwargs that the decorator can take, but your IDE should give you code completion and type hints.
    """

    container_name: Optional[str] = None
    args: Optional[List[str]] = None
    command: Optional[List[str]] = None
    lifecycle: Optional[Lifecycle] = None
    security_context: Optional[SecurityContext] = None
    source: Optional[Union[Callable, str]] = None
    working_dir: Optional[str] = None
    add_cwd_to_sys_path: Optional[bool] = None
    constructor: Optional[Union[str, ScriptConstructor]] = None

    @validator("constructor", always=True)
    @classmethod
    def _set_constructor(cls, v):
        if v is None:
            # TODO: In the future we can insert
            # detection code here to determine
            # the best constructor to use.
            v = InlineScriptConstructor()
        if isinstance(v, ScriptConstructor):
            return v
        assert isinstance(v, str)
        if v.lower() == "inline":
            return InlineScriptConstructor()
        elif v.lower() == "runner":
            return RunnerScriptConstructor()
        raise ValueError(f"Unknown constructor {v}")

    @validator("command", always=True)
    @classmethod
    def _set_command(cls, v):
        return v or global_config.script_command

    @validator("add_cwd_to_sys_path", always=True)
    @classmethod
    def _set_add_cwd_to_sys_path(cls, v):
        if v is None:
            return True

    @root_validator
    @classmethod
    def _constructor_validate(cls, values):
        constructor = values.get("constructor")
        assert isinstance(constructor, ScriptConstructor)
        return constructor.transform_values(cls, values)

    def _build_template(self) -> _ModelTemplate:
        assert isinstance(self.constructor, ScriptConstructor)

        return self.constructor.transform_template_post_build(
            self,
            _ModelTemplate(
                active_deadline_seconds=self.active_deadline_seconds,
                affinity=self.affinity,
                archive_location=self.archive_location,
                automount_service_account_token=self.automount_service_account_token,
                daemon=self.daemon,
                executor=self.executor,
                fail_fast=self.fail_fast,
                host_aliases=self.host_aliases,
                init_containers=self._build_init_containers(),
                inputs=self._build_inputs(),
                memoize=self.memoize,
                metadata=self._build_metadata(),
                metrics=self._build_metrics(),
                name=self.name,
                node_selector=self.node_selector,
                outputs=self._build_outputs(),
                parallelism=self.parallelism,
                plugin=self.plugin,
                pod_spec_patch=self.pod_spec_patch,
                priority=self.priority,
                priority_class_name=self.priority_class_name,
                retry_strategy=self.retry_strategy,
                scheduler_name=self.scheduler_name,
                script=self._build_script(),
                security_context=self.pod_security_context,
                service_account_name=self.service_account_name,
                sidecars=self._build_sidecars(),
                synchronization=self.synchronization,
                timeout=self.timeout,
                tolerations=self.tolerations,
                volumes=self._build_volumes(),
            ),
        )

    def _build_script(self) -> _ModelScriptTemplate:
        assert isinstance(self.constructor, ScriptConstructor)
        if _output_annotations_used(cast(Callable, self.source)) and isinstance(
            self.constructor, RunnerScriptConstructor
        ):
            if not self.constructor.outputs_directory:
                self.constructor.outputs_directory = self.constructor.DEFAULT_HERA_OUTPUTS_DIRECTORY
            if self.constructor.volume_for_outputs is not None:
                if self.constructor.volume_for_outputs.mount_path is None:
                    self.constructor.volume_for_outputs.mount_path = self.constructor.outputs_directory
                self._create_hera_outputs_volume(self.constructor.volume_for_outputs)
        assert self.image
        return self.constructor.transform_script_template_post_build(
            self,
            _ModelScriptTemplate(
                args=self.args,
                command=self.command,
                env=self._build_env(),
                env_from=self._build_env_from(),
                image=self.image,
                # `image_pull_policy` in script wants a string not an `ImagePullPolicy` object
                image_pull_policy=self._build_image_pull_policy(),
                lifecycle=self.lifecycle,
                liveness_probe=self.liveness_probe,
                name=self.container_name,
                ports=self.ports,
                readiness_probe=self.readiness_probe,
                resize_policy=self.resize_policy,
                resources=self._build_resources(),
                restart_policy=self.restart_policy,
                security_context=self.security_context,
                source=self.constructor.generate_source(self),
                startup_probe=self.startup_probe,
                stdin=self.stdin,
                stdin_once=self.stdin_once,
                termination_message_path=self.termination_message_path,
                termination_message_policy=self.termination_message_policy,
                tty=self.tty,
                volume_devices=self.volume_devices,
                volume_mounts=self._build_volume_mounts(),
                working_dir=self.working_dir,
            ),
        )

    def _build_inputs(self) -> Optional[ModelInputs]:
        inputs = super()._build_inputs()
        func_parameters: List[Parameter] = []
        func_artifacts: List[Artifact] = []
        if callable(self.source):
            func_parameters, func_artifacts = _get_inputs_from_callable(self.source)

        return cast(Optional[ModelInputs], self._aggregate_callable_io(inputs, func_parameters, func_artifacts, False))

    def _build_outputs(self) -> Optional[ModelOutputs]:
        outputs = super()._build_outputs()

        if not callable(self.source):
            return outputs

        outputs_directory = None
        if isinstance(self.constructor, RunnerScriptConstructor):
            outputs_directory = self.constructor.outputs_directory or self.constructor.DEFAULT_HERA_OUTPUTS_DIRECTORY

        out_parameters, out_artifacts = _get_outputs_from_return_annotation(self.source, outputs_directory)
        func_parameters, func_artifacts = _get_outputs_from_parameter_annotations(self.source, outputs_directory)
        func_parameters.extend(out_parameters)
        func_artifacts.extend(out_artifacts)

        return cast(
            Optional[ModelOutputs], self._aggregate_callable_io(outputs, func_parameters, func_artifacts, True)
        )

    def _aggregate_callable_io(
        self,
        current_io: Optional[Union[ModelInputs, ModelOutputs]],
        func_parameters: List[Parameter],
        func_artifacts: List[Artifact],
        output: bool,
    ) -> Union[ModelOutputs, ModelInputs, None]:
        """Aggregate the Inputs/Outputs with parameters and artifacts extracted from a callable."""
        if not func_parameters and not func_artifacts:
            return current_io
        if current_io is None:
            if output:
                return ModelOutputs(
                    parameters=[p.as_output() for p in func_parameters] or None,
                    artifacts=[a._build_artifact() for a in func_artifacts] or None,
                )

            return ModelInputs(
                parameters=[p.as_input() for p in func_parameters] or None,
                artifacts=[a._build_artifact() for a in func_artifacts] or None,
            )

        seen_params = {p.name for p in current_io.parameters or []}
        seen_artifacts = {a.name for a in current_io.artifacts or []}

        for param in func_parameters:
            if param.name not in seen_params and param.name not in seen_artifacts:
                if current_io.parameters is None:
                    current_io.parameters = []
                if output:
                    current_io.parameters.append(param.as_output())
                else:
                    current_io.parameters.append(param.as_input())

        for artifact in func_artifacts:
            if artifact.name not in seen_artifacts:
                if current_io.artifacts is None:
                    current_io.artifacts = []
                current_io.artifacts.append(artifact._build_artifact())

        return current_io

    def _create_hera_outputs_volume(self, volume: _BaseVolume) -> None:
        """Add given volume to the script template for the automatic saving of the hera outputs."""
        assert isinstance(self.constructor, RunnerScriptConstructor)

        if self.volumes is None:
            self.volumes = []
        elif isinstance(self.volumes, Sequence):
            self.volumes = list(self.volumes)
        elif not isinstance(self.volumes, list):
            self.volumes = [self.volumes]

        if volume not in self.volumes:
            self.volumes.append(volume)

active_deadline_seconds 

active_deadline_seconds: Optional[IntOrString] = None

add_cwd_to_sys_path 

add_cwd_to_sys_path: Optional[bool] = None

affinity 

affinity: Optional[Affinity] = None

annotations 

annotations: Optional[Dict[str, str]] = None

archive_location 

archive_location: Optional[ArtifactLocation] = None

args 

args: Optional[List[str]] = None

automount_service_account_token 

automount_service_account_token: Optional[bool] = None

command 

command: Optional[List[str]] = None

constructor 

constructor: Optional[Union[str, ScriptConstructor]] = None

container_name 

container_name: Optional[str] = None

daemon 

daemon: Optional[bool] = None

env 

env: EnvT = None

env_from 

env_from: EnvFromT = None

executor 

executor: Optional[ExecutorConfig] = None

fail_fast 

fail_fast: Optional[bool] = None

host_aliases 

host_aliases: Optional[List[HostAlias]] = None

http 

http: Optional[HTTP] = None

image 

image: Optional[str] = None

image_pull_policy 

image_pull_policy: Optional[Union[str, ImagePullPolicy]] = None

init_containers 

init_containers: Optional[List[Union[UserContainer, UserContainer]]] = None

inputs 

inputs: InputsT = None

labels 

labels: Optional[Dict[str, str]] = None

lifecycle 

lifecycle: Optional[Lifecycle] = None

liveness_probe 

liveness_probe: Optional[Probe] = None

memoize 

memoize: Optional[Memoize] = None

metrics 

metrics: Optional[MetricsT] = None

name 

name: Optional[str] = None

node_selector 

node_selector: Optional[Dict[str, str]] = None

outputs 

outputs: OutputsT = None

parallelism 

parallelism: Optional[int] = None

plugin 

plugin: Optional[Plugin] = None

pod_security_context 

pod_security_context: Optional[PodSecurityContext] = None

pod_spec_patch 

pod_spec_patch: Optional[str] = None

ports 

ports: Optional[List[ContainerPort]] = None

priority 

priority: Optional[int] = None

priority_class_name 

priority_class_name: Optional[str] = None

readiness_probe 

readiness_probe: Optional[Probe] = None

resize_policy 

resize_policy: Optional[List[ContainerResizePolicy]] = None

resources 

resources: Optional[Union[ResourceRequirements, Resources]] = None

restart_policy 

restart_policy: Optional[str] = None

retry_strategy 

retry_strategy: Optional[RetryStrategy] = None

scheduler_name 

scheduler_name: Optional[str] = None

security_context 

security_context: Optional[SecurityContext] = None

service_account_name 

service_account_name: Optional[str] = None

sidecars 

sidecars: Optional[Union[UserContainer, List[UserContainer]]] = None

source 

source: Optional[Union[Callable, str]] = None

startup_probe 

startup_probe: Optional[Probe] = None

stdin 

stdin: Optional[bool] = None

stdin_once 

stdin_once: Optional[bool] = None

synchronization 

synchronization: Optional[Synchronization] = None

termination_message_path 

termination_message_path: Optional[str] = None

termination_message_policy 

termination_message_policy: Optional[str] = None

timeout 

timeout: Optional[str] = None

tolerations 

tolerations: Optional[List[Toleration]] = None

tty 

tty: Optional[bool] = None

volume_devices 

volume_devices: Optional[List[VolumeDevice]] = None

volume_mounts 

volume_mounts: Optional[List[VolumeMount]] = None

volumes 

volumes: Optional[VolumesT] = None

working_dir 

working_dir: Optional[str] = None

get_artifact 

get_artifact(name: str) -> Artifact

Finds and returns the artifact with the supplied name.

Note that this method will raise an error if the artifact is not found.

Parameters:

Name Type Description Default
name str

name of the input artifact to find and return.

 required

Returns:

Name Type Description
Artifact Artifact

the artifact with the supplied name.

Raises:

Type Description
KeyError

if the artifact is not found.
Source code in src/hera/workflows/_mixins.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def get_artifact(self, name: str) -> Artifact:
    """Finds and returns the artifact with the supplied name.

    Note that this method will raise an error if the artifact is not found.

    Args:
        name: name of the input artifact to find and return.

    Returns:
        Artifact: the artifact with the supplied name.

    Raises:
        KeyError: if the artifact is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Artifact {name} not found.")
    if inputs.artifacts is None:
        raise KeyError(f"No artifacts set. Artifact {name} not found.")
    for artifact in inputs.artifacts:
        if artifact.name == name:
            return Artifact(name=name, from_=f"{{{{inputs.artifacts.{artifact.name}}}}}")
    raise KeyError(f"Artifact {name} not found.")

get_parameter 

get_parameter(name: str) -> Parameter

Finds and returns the parameter with the supplied name.

Note that this method will raise an error if the parameter is not found.

Parameters:

Name Type Description Default
name str

name of the input parameter to find and return.

 required

Returns:

Name Type Description
Parameter Parameter

the parameter with the supplied name.

Raises:

Type Description
KeyError

if the parameter is not found.
Source code in src/hera/workflows/_mixins.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def get_parameter(self, name: str) -> Parameter:
    """Finds and returns the parameter with the supplied name.

    Note that this method will raise an error if the parameter is not found.

    Args:
        name: name of the input parameter to find and return.

    Returns:
        Parameter: the parameter with the supplied name.

    Raises:
        KeyError: if the parameter is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Parameter {name} not found.")
    if inputs.parameters is None:
        raise KeyError(f"No parameters set. Parameter {name} not found.")
    for p in inputs.parameters:
        if p.name == name:
            param = Parameter.from_model(p)
            param.value = f"{{{{inputs.parameters.{param.name}}}}}"
            return param
    raise KeyError(f"Parameter {name} not found.")

Steps

A Steps template invocator is used to define a sequence of steps which can run sequentially or in parallel.

Steps implements the contextmanager interface so allows usage of with, under which any hera.workflows.steps.Step objects instantiated will be added to the Steps’ list of sub_steps.

  • Step and Parallel objects initialised within a Steps context will be added to the list of sub_steps in the order they are initialised.
  • All Step objects initialised within a Parallel context will run in parallel.
Source code in src/hera/workflows/steps.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
class Steps(
    IOMixin,
    TemplateMixin,
    CallableTemplateMixin,
    ContextMixin,
):
    """A Steps template invocator is used to define a sequence of steps which can run sequentially or in parallel.

    Steps implements the contextmanager interface so allows usage of `with`, under which any
    `hera.workflows.steps.Step` objects instantiated will be added to the Steps' list of sub_steps.

    * Step and Parallel objects initialised within a Steps context will be added to the list of sub_steps
    in the order they are initialised.
    * All Step objects initialised within a Parallel context will run in parallel.
    """

    _node_names = PrivateAttr(default_factory=set)

    sub_steps: List[
        Union[
            Step,
            Parallel,
            List[Step],
            _ModelWorkflowStep,
            List[_ModelWorkflowStep],
        ]
    ] = []

    def _build_steps(self) -> Optional[List[ParallelSteps]]:
        steps = []
        for workflow_step in self.sub_steps:
            if isinstance(workflow_step, Steppable):
                steps.append(ParallelSteps(__root__=workflow_step._build_step()))
            elif isinstance(workflow_step, _ModelWorkflowStep):
                steps.append(ParallelSteps(__root__=[workflow_step]))
            elif isinstance(workflow_step, List):
                substeps = []
                for s in workflow_step:
                    if isinstance(s, Step):
                        substeps.append(s._build_as_workflow_step())
                    elif isinstance(s, _ModelWorkflowStep):
                        substeps.append(s)
                    else:
                        raise InvalidType(type(s))
                steps.append(ParallelSteps(__root__=substeps))
            else:
                raise InvalidType(type(workflow_step))

        return steps or None

    def _add_sub(self, node: Any):
        if not isinstance(node, (Step, Parallel)):
            raise InvalidType(type(node))
        if isinstance(node, Step):
            if node.name in self._node_names:
                raise NodeNameConflict(f"Found multiple Step nodes with name: {node.name}")
            self._node_names.add(node.name)
        if isinstance(node, Parallel):
            node._node_names = self._node_names
        self.sub_steps.append(node)

    def parallel(self) -> Parallel:
        """Returns a Parallel object which can be used in a sub-context manager."""
        return Parallel()

    def _build_template(self) -> _ModelTemplate:
        return _ModelTemplate(
            active_deadline_seconds=self.active_deadline_seconds,
            affinity=self.affinity,
            archive_location=self.archive_location,
            automount_service_account_token=self.automount_service_account_token,
            container=None,
            container_set=None,
            daemon=self.daemon,
            dag=None,
            data=None,
            executor=self.executor,
            fail_fast=self.fail_fast,
            http=None,
            host_aliases=self.host_aliases,
            init_containers=self._build_init_containers(),
            inputs=self._build_inputs(),
            memoize=self.memoize,
            metadata=self._build_metadata(),
            metrics=self._build_metrics(),
            name=self.name,
            node_selector=self.node_selector,
            outputs=self._build_outputs(),
            parallelism=self.parallelism,
            plugin=self.plugin,
            pod_spec_patch=self.pod_spec_patch,
            priority=self.priority,
            priority_class_name=self.priority_class_name,
            resource=None,
            retry_strategy=self.retry_strategy,
            scheduler_name=self.scheduler_name,
            script=None,
            security_context=self.pod_security_context,
            service_account_name=self.service_account_name,
            sidecars=self._build_sidecars(),
            steps=self._build_steps(),
            suspend=None,
            synchronization=self.synchronization,
            timeout=self.timeout,
            tolerations=self.tolerations,
        )

active_deadline_seconds 

active_deadline_seconds: Optional[IntOrString] = None

affinity 

affinity: Optional[Affinity] = None

annotations 

annotations: Optional[Dict[str, str]] = None

archive_location 

archive_location: Optional[ArtifactLocation] = None

automount_service_account_token 

automount_service_account_token: Optional[bool] = None

daemon 

daemon: Optional[bool] = None

executor 

executor: Optional[ExecutorConfig] = None

fail_fast 

fail_fast: Optional[bool] = None

host_aliases 

host_aliases: Optional[List[HostAlias]] = None

http 

http: Optional[HTTP] = None

init_containers 

init_containers: Optional[List[Union[UserContainer, UserContainer]]] = None

inputs 

inputs: InputsT = None

labels 

labels: Optional[Dict[str, str]] = None

memoize 

memoize: Optional[Memoize] = None

metrics 

metrics: Optional[MetricsT] = None

name 

name: Optional[str] = None

node_selector 

node_selector: Optional[Dict[str, str]] = None

outputs 

outputs: OutputsT = None

parallelism 

parallelism: Optional[int] = None

plugin 

plugin: Optional[Plugin] = None

pod_security_context 

pod_security_context: Optional[PodSecurityContext] = None

pod_spec_patch 

pod_spec_patch: Optional[str] = None

priority 

priority: Optional[int] = None

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[RetryStrategy] = None

scheduler_name 

scheduler_name: Optional[str] = None

service_account_name 

service_account_name: Optional[str] = None

sidecars 

sidecars: Optional[Union[UserContainer, List[UserContainer]]] = None

sub_steps 

sub_steps: List[Union[Step, Parallel, List[Step], WorkflowStep, List[WorkflowStep]]] = []

synchronization 

synchronization: Optional[Synchronization] = None

timeout 

timeout: Optional[str] = None

tolerations 

tolerations: Optional[List[Toleration]] = None

get_artifact 

get_artifact(name: str) -> Artifact

Finds and returns the artifact with the supplied name.

Note that this method will raise an error if the artifact is not found.

Parameters:

Name Type Description Default
name str

name of the input artifact to find and return.

 required

Returns:

Name Type Description
Artifact Artifact

the artifact with the supplied name.

Raises:

Type Description
KeyError

if the artifact is not found.
Source code in src/hera/workflows/_mixins.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def get_artifact(self, name: str) -> Artifact:
    """Finds and returns the artifact with the supplied name.

    Note that this method will raise an error if the artifact is not found.

    Args:
        name: name of the input artifact to find and return.

    Returns:
        Artifact: the artifact with the supplied name.

    Raises:
        KeyError: if the artifact is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Artifact {name} not found.")
    if inputs.artifacts is None:
        raise KeyError(f"No artifacts set. Artifact {name} not found.")
    for artifact in inputs.artifacts:
        if artifact.name == name:
            return Artifact(name=name, from_=f"{{{{inputs.artifacts.{artifact.name}}}}}")
    raise KeyError(f"Artifact {name} not found.")

get_parameter 

get_parameter(name: str) -> Parameter

Finds and returns the parameter with the supplied name.

Note that this method will raise an error if the parameter is not found.

Parameters:

Name Type Description Default
name str

name of the input parameter to find and return.

 required

Returns:

Name Type Description
Parameter Parameter

the parameter with the supplied name.

Raises:

Type Description
KeyError

if the parameter is not found.
Source code in src/hera/workflows/_mixins.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def get_parameter(self, name: str) -> Parameter:
    """Finds and returns the parameter with the supplied name.

    Note that this method will raise an error if the parameter is not found.

    Args:
        name: name of the input parameter to find and return.

    Returns:
        Parameter: the parameter with the supplied name.

    Raises:
        KeyError: if the parameter is not found.
    """
    inputs = self._build_inputs()
    if inputs is None:
        raise KeyError(f"No inputs set. Parameter {name} not found.")
    if inputs.parameters is None:
        raise KeyError(f"No parameters set. Parameter {name} not found.")
    for p in inputs.parameters:
        if p.name == name:
            param = Parameter.from_model(p)
            param.value = f"{{{{inputs.parameters.{param.name}}}}}"
            return param
    raise KeyError(f"Parameter {name} not found.")

parallel 

parallel() -> Parallel

Returns a Parallel object which can be used in a sub-context manager.

Source code in src/hera/workflows/steps.py 
194
195
196
def parallel(self) -> Parallel:
    """Returns a Parallel object which can be used in a sub-context manager."""
    return Parallel()

add_template_set 

add_template_set(template_set: TemplateSet) -> None

Add the templates stored in the template_set to this Workflow.

Source code in src/hera/workflows/workflow.py 
529
530
531
532
def add_template_set(self, template_set: TemplateSet) -> None:
    """Add the templates stored in the template_set to this Workflow."""
    for template in template_set.templates:
        self._add_sub(template)

build 

build() -> TWorkflow

Builds the ClusterWorkflowTemplate and its components into an Argo schema ClusterWorkflowTemplate object.

Source code in src/hera/workflows/cluster_workflow_template.py 
69
70
71
72
73
74
75
def build(self) -> TWorkflow:
    """Builds the ClusterWorkflowTemplate and its components into an Argo schema ClusterWorkflowTemplate object."""
    # Note that ClusterWorkflowTemplates are exactly the same as WorkflowTemplates except for the kind which is
    # handled in Workflow._set_kind (by __name__). When using ClusterWorkflowTemplates via templateRef, clients
    # should specify cluster_scope=True, but that is an intrinsic property of ClusterWorkflowTemplates from our
    # perspective.
    return _ModelClusterWorkflowTemplate(**super().build().dict())

container 

container(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
@_add_type_hints(Container)  # type: ignore
def container(self, **container_kwargs) -> Callable:
    self._check_if_enabled("container")
    from hera.workflows.container import Container

    def container_decorator(func: Callable[FuncIns, FuncR]) -> Callable:
        name = container_kwargs.pop("name", func.__name__.replace("_", "-"))

        signature = inspect.signature(func)
        func_inputs = signature.parameters
        inputs = []
        if len(func_inputs) >= 1:
            input_arg = list(func_inputs.values())[0].annotation
            if issubclass(input_arg, (InputV1, InputV2)):
                inputs = input_arg._get_inputs(add_missing_path=True)

        func_return = signature.return_annotation
        outputs = []
        if func_return and issubclass(func_return, (OutputV1, OutputV2)):
            outputs = func_return._get_outputs(add_missing_path=True)

        # Open context to add `Container` object automatically
        with self:
            container_template = Container(
                name=name,
                inputs=inputs,
                outputs=outputs,
                **container_kwargs,
            )

        @functools.wraps(func)
        def container_call_wrapper(*args, **kwargs) -> Union[FuncR, Step, Task, None]:
            """Invokes a CallableTemplateMixin's `__call__` method using the given SubNode (Step or Task) args/kwargs."""
            if _context.declaring:
                subnode_name = kwargs.pop("name", None)
                if not subnode_name:
                    try:
                        # ignore decorator function assignment
                        subnode_name = varname()
                    except ImproperUseError:
                        # Template is being used without variable assignment (so use function name or provided name)
                        subnode_name = container_template.name  # type: ignore

                    assert isinstance(subnode_name, str)
                    subnode_name = subnode_name.replace("_", "-")

                return self._create_subnode(subnode_name, func, container_template, *args, **kwargs)

            if _context.pieces:
                return container_template.__call__(*args, **kwargs)

            # Do not allow kwargs
            return func(*args)  # type: ignore

        # Set the template name to the inferred name
        container_call_wrapper.template_name = name  # type: ignore

        return container_call_wrapper

    return container_decorator

create 

create() -> TWorkflow

Creates the ClusterWorkflowTemplate on the Argo cluster.

Source code in src/hera/workflows/cluster_workflow_template.py 
26
27
28
29
30
31
def create(self) -> TWorkflow:  # type: ignore
    """Creates the ClusterWorkflowTemplate on the Argo cluster."""
    assert self.workflows_service, "workflow service not initialized"
    return self.workflows_service.create_cluster_workflow_template(
        ClusterWorkflowTemplateCreateRequest(template=self.build())  # type: ignore
    )

create_as_workflow 

create_as_workflow(generate_name: Optional[str] = None, wait: bool = False, poll_interval: int = 5) -> TWorkflow

Run this WorkflowTemplate instantly as a Workflow.

If generate_name is given, the workflow created uses generate_name as a prefix, as per the usual for hera.workflows.Workflow.generate_name. If not given, the WorkflowTemplate’s name will be used, truncated to 57 chars and appended with a hyphen.

Note: this function does not require the WorkflowTemplate to already exist on the cluster

Source code in src/hera/workflows/workflow_template.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def create_as_workflow(
    self,
    generate_name: Optional[str] = None,
    wait: bool = False,
    poll_interval: int = 5,
) -> TWorkflow:
    """Run this WorkflowTemplate instantly as a Workflow.

    If generate_name is given, the workflow created uses generate_name as a prefix, as per the usual for
    hera.workflows.Workflow.generate_name. If not given, the WorkflowTemplate's name will be used, truncated to 57
    chars and appended with a hyphen.

    Note: this function does not require the WorkflowTemplate to already exist on the cluster
    """
    workflow = self._get_as_workflow(generate_name)
    return workflow.create(wait=wait, poll_interval=poll_interval)

dag 

dag(**dag_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
961
962
963
964
965
966
967
@_add_type_hints(DAG)  # type: ignore
def dag(self, **dag_kwargs) -> Callable:
    self._check_if_enabled("dag")

    from hera.workflows.dag import DAG

    return self._construct_invocator_decorator(DAG, **dag_kwargs)

from_dict 

from_dict(model_dict: Dict) -> ModelMapperMixin

Create a WorkflowTemplate from a WorkflowTemplate contained in a dict.

Examples:

>>> my_workflow_template = WorkflowTemplate(name="my-wft")
>>> my_workflow_template == WorkflowTemplate.from_dict(my_workflow_template.to_dict())
True
Source code in src/hera/workflows/workflow_template.py 
118
119
120
121
122
123
124
125
126
127
@classmethod
def from_dict(cls, model_dict: Dict) -> ModelMapperMixin:
    """Create a WorkflowTemplate from a WorkflowTemplate contained in a dict.

    Examples:
        >>> my_workflow_template = WorkflowTemplate(name="my-wft")
        >>> my_workflow_template == WorkflowTemplate.from_dict(my_workflow_template.to_dict())
        True
    """
    return cls._from_dict(model_dict, _ModelWorkflowTemplate)

from_file 

from_file(yaml_file: Union[Path, str]) -> ModelMapperMixin

Create a WorkflowTemplate from a WorkflowTemplate contained in a YAML file.

Examples:

>>> yaml_file = Path(...)
>>> my_workflow_template = WorkflowTemplate.from_file(yaml_file)
Source code in src/hera/workflows/workflow_template.py 
138
139
140
141
142
143
144
145
146
@classmethod
def from_file(cls, yaml_file: Union[Path, str]) -> ModelMapperMixin:
    """Create a WorkflowTemplate from a WorkflowTemplate contained in a YAML file.

    Examples:
        >>> yaml_file = Path(...)
        >>> my_workflow_template = WorkflowTemplate.from_file(yaml_file)
    """
    return cls._from_file(yaml_file, _ModelWorkflowTemplate)

from_yaml 

from_yaml(yaml_str: str) -> ModelMapperMixin

Create a WorkflowTemplate from a WorkflowTemplate contained in a YAML string.

Examples:

>>> my_workflow_template = WorkflowTemplate.from_yaml(yaml_str)
Source code in src/hera/workflows/workflow_template.py 
129
130
131
132
133
134
135
136
@classmethod
def from_yaml(cls, yaml_str: str) -> ModelMapperMixin:
    """Create a WorkflowTemplate from a WorkflowTemplate contained in a YAML string.

    Examples:
        >>> my_workflow_template = WorkflowTemplate.from_yaml(yaml_str)
    """
    return cls._from_yaml(yaml_str, _ModelWorkflowTemplate)

get 

get() -> TWorkflow

Attempts to get a workflow template based on the parameters of this template e.g. name + namespace.

Source code in src/hera/workflows/cluster_workflow_template.py 
33
34
35
36
37
def get(self) -> TWorkflow:
    """Attempts to get a workflow template based on the parameters of this template e.g. name + namespace."""
    assert self.workflows_service, "workflow service not initialized"
    assert self.name, "workflow name not defined"
    return self.workflows_service.get_cluster_workflow_template(name=self.name)

get_parameter 

get_parameter(name: str) -> Parameter

Attempts to find and return a Parameter of the specified name.

Source code in src/hera/workflows/workflow.py 
310
311
312
313
314
315
316
317
318
319
320
321
def get_parameter(self, name: str) -> Parameter:
    """Attempts to find and return a `Parameter` of the specified name."""
    arguments = self._build_arguments()
    if arguments is None:
        raise KeyError("Workflow has no arguments set")
    if arguments.parameters is None:
        raise KeyError("Workflow has no argument parameters set")

    parameters = arguments.parameters
    if next((p for p in parameters if p.name == name), None) is None:
        raise KeyError(f"`{name}` is not a valid workflow parameter")
    return Parameter(name=name, value=f"{{{{workflow.parameters.{name}}}}}")

get_workflow_link() -> str

Returns the workflow link for the workflow.

Source code in src/hera/workflows/workflow.py 
509
510
511
512
513
def get_workflow_link(self) -> str:
    """Returns the workflow link for the workflow."""
    assert self.workflows_service is not None, "Cannot fetch a workflow link without a service"
    assert self.name is not None, "Cannot fetch a workflow link without a workflow name"
    return self.workflows_service.get_workflow_link(self.name)

lint 

lint() -> TWorkflow

Lints the ClusterWorkflowTemplate using the Argo cluster.

Source code in src/hera/workflows/cluster_workflow_template.py 
62
63
64
65
66
67
def lint(self) -> TWorkflow:
    """Lints the ClusterWorkflowTemplate using the Argo cluster."""
    assert self.workflows_service, "workflow service not initialized"
    return self.workflows_service.lint_cluster_workflow_template(
        ClusterWorkflowTemplateLintRequest(template=self.build())  # type: ignore
    )

script 

script(**script_kwargs) -> Callable

A decorator that wraps a function into a Script object.

Using this decorator users can define a function that will be executed as a script in a container. Once the Script is returned users can use it as they generally use a Script e.g. as a callable inside a DAG or Steps. Note that invoking the function will result in the template associated with the script to be added to the workflow context, so users do not have to worry about that.

Parameters:

Name Type Description Default
**script_kwargs

Keyword arguments to be passed to the Script object.

 {}

Returns:

Type Description
Callable

Function wrapper that holds a Script and allows the function to be called to create a Step or Task if in a Steps or DAG context.
Source code in src/hera/workflows/_meta_mixins.py 
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
@_add_type_hints(Script)  # type: ignore
def script(self, **script_kwargs) -> Callable:
    """A decorator that wraps a function into a Script object.

    Using this decorator users can define a function that will be executed as a script in a container. Once the
    `Script` is returned users can use it as they generally use a `Script` e.g. as a callable inside a DAG or Steps.
    Note that invoking the function will result in the template associated with the script to be added to the
    workflow context, so users do not have to worry about that.

    Args:
        **script_kwargs: Keyword arguments to be passed to the Script object.

    Returns:
        Function wrapper that holds a `Script` and allows the function to be called to create a Step or Task if in a Steps or DAG context.
    """
    self._check_if_enabled("script")

    from hera.workflows.script import RunnerScriptConstructor, Script

    def script_decorator(func: Callable[FuncIns, FuncR]) -> Callable:
        """The internal decorator function.

        Parameters
        ----------
        func: Callable
            Function to wrap.

        Returns:
        -------
        Callable
            Callable that calls the `Script` object `__call__` method when in a Steps or DAG context,
            otherwise calls function itself.
        """
        # instance methods are wrapped in `staticmethod`. Hera can capture that type and extract the underlying
        # function for remote submission since it does not depend on any class or instance attributes, so it is
        # submittable
        if isinstance(func, staticmethod):
            source: Callable = func.__func__
        else:
            source = func

        # take the client-provided `name` if it is submitted, pop the name for otherwise there will be two
        # kwargs called `name`
        # otherwise populate the `name` from the function name
        name = script_kwargs.pop("name", source.__name__.replace("_", "-"))

        if "constructor" not in script_kwargs and "constructor" not in global_config._get_class_defaults(Script):
            script_kwargs["constructor"] = RunnerScriptConstructor()

        signature = inspect.signature(func)
        func_inputs = signature.parameters
        if len(func_inputs) > 1:
            raise SyntaxError("script decorator must be used with a single `Input` arg, or no args.")

        if len(func_inputs) == 1:
            func_input = list(func_inputs.values())[0].annotation
            if not issubclass(func_input, (InputV1, InputV2)):
                raise SyntaxError("script decorator must be used with a single `Input` arg, or no args.")

        # Open (Workflow) context to add `Script` object automatically
        with self:
            script_template = Script(name=name, source=source, **script_kwargs)

        if not isinstance(script_template.constructor, RunnerScriptConstructor):
            raise ValueError(f"Script '{name}' must use RunnerScriptConstructor")

        @functools.wraps(func)
        def script_call_wrapper(*args, **kwargs) -> Union[FuncR, Step, Task, None]:
            """Invokes a CallableTemplateMixin's `__call__` method using the given SubNode (Step or Task) args/kwargs."""
            if _context.declaring:
                subnode_name = kwargs.pop("name", None)
                if not subnode_name:
                    try:
                        # ignore decorator function assignment
                        subnode_name = varname()
                    except ImproperUseError:
                        # Template is being used without variable assignment (so use function name or provided name)
                        subnode_name = script_template.name  # type: ignore

                    assert isinstance(subnode_name, str)
                    subnode_name = subnode_name.replace("_", "-")

                return self._create_subnode(subnode_name, func, script_template, *args, **kwargs)

            if _context.pieces:
                return script_template.__call__(*args, **kwargs)

            # Do not allow kwargs
            return func(*args)  # type: ignore

        # Set the wrapped function to the original function so that we can use it later
        script_call_wrapper.wrapped_function = func  # type: ignore
        # Set the template name to the inferred name
        script_call_wrapper.template_name = name  # type: ignore

        return script_call_wrapper

    return script_decorator

set_entrypoint 

set_entrypoint(func: Callable[P, T]) -> Callable[P, T]

Decorator function to set entrypoint.

Source code in src/hera/workflows/workflow.py 
515
516
517
518
519
520
521
522
523
524
525
526
527
def set_entrypoint(self, func: Callable[P, T]) -> Callable[P, T]:
    """Decorator function to set entrypoint."""
    if not hasattr(func, "template_name"):
        raise SyntaxError("`set_entrypoint` decorator must be above template decorator")

    if self.entrypoint is not None:
        if self.entrypoint == func.template_name:
            return func

        logging.warning(f"entrypoint is being reassigned from {self.entrypoint} to {func.template_name}")

    self.entrypoint = func.template_name  # type: ignore
    return func

steps 

steps(**steps_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
969
970
971
972
973
974
975
@_add_type_hints(Steps)  # type: ignore
def steps(self, **steps_kwargs) -> Callable:
    self._check_if_enabled("steps")

    from hera.workflows.steps import Steps

    return self._construct_invocator_decorator(Steps, **steps_kwargs)

to_dict 

to_dict() -> Any

Builds the Workflow as an Argo schema Workflow object and returns it as a dictionary.

Source code in src/hera/workflows/workflow.py 
333
334
335
def to_dict(self) -> Any:
    """Builds the Workflow as an Argo schema Workflow object and returns it as a dictionary."""
    return self.build().dict(exclude_none=True, by_alias=True)

to_file 

to_file(output_directory: Union[Path, str] = '.', name: str = '', *args, **kwargs) -> Path

Writes the Workflow as an Argo schema Workflow object to a YAML file and returns the path to the file.

Parameters:

Name Type Description Default
output_directory Union[Path, str]

The directory to write the file to. Defaults to the current working directory.

 '.'
name str

The name of the file to write without the file extension. Defaults to the Workflow’s name or a generated name.

 ''
*args

Additional arguments to pass to yaml.dump.

 ()
**kwargs

Additional keyword arguments to pass to yaml.dump.

 {}
Source code in src/hera/workflows/workflow.py 
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
def to_file(self, output_directory: Union[Path, str] = ".", name: str = "", *args, **kwargs) -> Path:
    """Writes the Workflow as an Argo schema Workflow object to a YAML file and returns the path to the file.

    Args:
        output_directory: The directory to write the file to. Defaults to the current working directory.
        name: The name of the file to write without the file extension.  Defaults to the Workflow's name or a
              generated name.
        *args: Additional arguments to pass to `yaml.dump`.
        **kwargs: Additional keyword arguments to pass to `yaml.dump`.
    """
    workflow_name = self.name or (self.generate_name or "workflow").rstrip("-")
    name = name or workflow_name
    output_directory = Path(output_directory)
    output_path = Path(output_directory) / f"{name}.yaml"
    output_directory.mkdir(parents=True, exist_ok=True)
    output_path.write_text(self.to_yaml(*args, **kwargs))
    return output_path.absolute()

to_yaml 

to_yaml(*args, **kwargs) -> str

Builds the Workflow as an Argo schema Workflow object and returns it as yaml string.

Source code in src/hera/workflows/workflow.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def to_yaml(self, *args, **kwargs) -> str:
    """Builds the Workflow as an Argo schema Workflow object and returns it as yaml string."""

    def human_readable_ordering(kv: tuple) -> int:
        """Key ordering function for ordering in a more human-readable fashion.

        Ordering is:
        1. "name" keys always first (if present)
        2. Primitives (not dicts/lists)
        3. lists
        4. dict
        """
        k, v = kv
        if k == "name" and isinstance(v, str):
            return 0
        if not isinstance(v, (dict, list)):
            return 1
        if isinstance(v, list):
            return 2
        return 3

    def order_dict(d: dict) -> dict[str, Any]:
        """Recursively orders `d` by the custom_ordering function by inserting them into a copy of the dict in order."""
        d_copy: dict[str, Any] = dict()
        for k, v in sorted(d.items(), key=lambda x: (human_readable_ordering(x), x[0])):
            if isinstance(v, dict):
                d_copy[k] = order_dict(v)
            elif isinstance(v, list):
                if v and isinstance(v[0], dict):
                    d_copy[k] = [order_dict(i) if isinstance(i, dict) else i for i in v]
                elif v and isinstance(v[0], list):
                    d_copy[k] = [[order_dict(i) for i in sublist] for sublist in v]
                else:
                    d_copy[k] = v
            else:
                d_copy[k] = v
        return d_copy

    human_ordered_dict = order_dict(self.to_dict())
    return _yaml.dump(human_ordered_dict, *args, **kwargs)

update 

update() -> TWorkflow

Attempts to perform a workflow template update based on the parameters of this template.

Note that this creates the template if it does not exist. In addition, this performs a get prior to updating to get the resource version to update in the first place. If you know the template does not exist ahead of time, it is more efficient to use create() directly to avoid one round trip.

Source code in src/hera/workflows/cluster_workflow_template.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def update(self) -> TWorkflow:
    """Attempts to perform a workflow template update based on the parameters of this template.

    Note that this creates the template if it does not exist. In addition, this performs
    a get prior to updating to get the resource version to update in the first place. If you know the template
    does not exist ahead of time, it is more efficient to use `create()` directly to avoid one round trip.
    """
    assert self.workflows_service, "workflow service not initialized"
    assert self.name, "workflow name not defined"
    # we always need to do a get prior to updating to get the resource version to update in the first place
    # https://github.com/argoproj/argo-workflows/pull/5465#discussion_r597797052

    template = self.build()
    try:
        curr = self.get()
        template.metadata.resource_version = curr.metadata.resource_version
    except NotFound:
        return self.create()
    return self.workflows_service.update_cluster_workflow_template(
        self.name,
        ClusterWorkflowTemplateUpdateRequest(template=template),  # type: ignore
    )

wait 

wait(poll_interval: int = 5) -> TWorkflow

Waits for the Workflow to complete execution.

Parameters

poll_interval: int = 5 The interval in seconds to poll the workflow status.

Source code in src/hera/workflows/workflow.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def wait(self, poll_interval: int = 5) -> TWorkflow:
    """Waits for the Workflow to complete execution.

    Parameters
    ----------
    poll_interval: int = 5
        The interval in seconds to poll the workflow status.
    """
    assert self.workflows_service is not None, "workflow service not initialized"
    assert self.namespace is not None, "workflow namespace not defined"
    assert self.name is not None, "workflow name not defined"

    # here we use the sleep interval to wait for the workflow post creation. This is to address a potential
    # race conditions such as:
    # 1. Argo server says "workflow was accepted" but the workflow is not yet created
    # 2. Hera wants to verify the status of the workflow, but it's not yet defined because it's not created
    # 3. Argo finally creates the workflow
    # 4. Hera throws an `AssertionError` because the phase assertion fails
    time.sleep(poll_interval)
    wf = self.workflows_service.get_workflow(self.name, namespace=self.namespace)
    assert wf.metadata.name is not None, f"workflow name not defined for workflow {self.name}"

    assert wf.status is not None, f"workflow status not defined for workflow {wf.metadata.name}"
    assert wf.status.phase is not None, f"workflow phase not defined for workflow status {wf.status}"
    status = WorkflowStatus.from_argo_status(wf.status.phase)

    # keep polling for workflow status until completed, at the interval dictated by the user
    while status == WorkflowStatus.running:
        time.sleep(poll_interval)
        wf = self.workflows_service.get_workflow(wf.metadata.name, namespace=self.namespace)
        assert wf.status is not None, f"workflow status not defined for workflow {wf.metadata.name}"
        assert wf.status.phase is not None, f"workflow phase not defined for workflow status {wf.status}"
        status = WorkflowStatus.from_argo_status(wf.status.phase)
    return wf

Comments