Skip to content

Cron workflow

CronWorkflow

CronWorkflow allows a user to run a Workflow on a recurring basis.

Note

Hera’s CronWorkflow is a subclass of Workflow which means certain fields are renamed for compatibility, see cron_suspend and cron_status which are different from the Argo spec. See CronWorkflowSpec for more details.

Source code in src/hera/workflows/cron_workflow.py 
 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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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
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
class CronWorkflow(Workflow):
    """CronWorkflow allows a user to run a Workflow on a recurring basis.

    Note:
        Hera's CronWorkflow is a subclass of Workflow which means certain fields are renamed
        for compatibility, see `cron_suspend` and `cron_status` which are different from the Argo
        spec. See [CronWorkflowSpec](https://argoproj.github.io/argo-workflows/fields/#cronworkflow) for more details.
    """

    concurrency_policy: Annotated[Optional[str], _CronWorkflowModelMapper("spec.concurrency_policy")] = None
    failed_jobs_history_limit: Annotated[Optional[int], _CronWorkflowModelMapper("spec.failed_jobs_history_limit")] = (
        None
    )
    schedule: Annotated[Optional[str], _CronWorkflowModelMapper("spec.schedule")] = None
    schedules: Annotated[Optional[List[str]], _CronWorkflowModelMapper("spec.schedules")] = None
    starting_deadline_seconds: Annotated[Optional[int], _CronWorkflowModelMapper("spec.starting_deadline_seconds")] = (
        None
    )
    stop_strategy: Annotated[Optional[StopStrategy], _CronWorkflowModelMapper("spec.stop_strategy")] = None
    successful_jobs_history_limit: Annotated[
        Optional[int], _CronWorkflowModelMapper("spec.successful_jobs_history_limit")
    ] = None
    cron_suspend: Annotated[Optional[bool], _CronWorkflowModelMapper("spec.suspend")] = None
    timezone: Annotated[Optional[str], _CronWorkflowModelMapper("spec.timezone")] = None
    when: Annotated[Optional[str], _CronWorkflowModelMapper("spec.when")] = None
    cron_status: Annotated[Optional[CronWorkflowStatus], _CronWorkflowModelMapper("status")] = None

    def create(self) -> TWorkflow:  # type: ignore
        """Creates the CronWorkflow on the Argo cluster."""
        assert isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
        assert self.namespace, "workflow namespace not defined"

        wf = self.workflows_service.create_cron_workflow(
            CreateCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
            namespace=self.namespace,
        )
        # set the name on the object so that we can do a get/update later
        self.name = wf.metadata.name
        return wf

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

    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 isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
        assert self.namespace, "workflow namespace not defined"
        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_cron_workflow(
            self.name,
            UpdateCronWorkflowRequest(cron_workflow=template),  # type: ignore
            namespace=self.namespace,
        )

    def lint(self) -> TWorkflow:
        """Lints the CronWorkflow using the Argo cluster."""
        assert isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
        assert self.namespace, "workflow namespace not defined"
        return self.workflows_service.lint_cron_workflow(
            LintCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
            namespace=self.namespace,
        )

    async def async_create(self) -> TWorkflow:  # type: ignore
        """Creates the CronWorkflow on the Argo cluster."""
        assert isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
        assert self.namespace, "workflow namespace not defined"

        wf = await self.workflows_service.create_cron_workflow(
            CreateCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
            namespace=self.namespace,
        )
        # set the name on the object so that we can do a get/update later
        self.name = wf.metadata.name
        return wf

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

    async def async_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 isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
        assert self.namespace, "workflow namespace not defined"
        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 = await self.async_get()
            template.metadata.resource_version = curr.metadata.resource_version
        except NotFound:
            return await self.async_create()
        return await self.workflows_service.update_cron_workflow(
            self.name,
            UpdateCronWorkflowRequest(cron_workflow=template),  # type: ignore
            namespace=self.namespace,
        )

    async def async_lint(self) -> TWorkflow:
        """Lints the CronWorkflow using the Argo cluster."""
        assert isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
        assert self.namespace, "workflow namespace not defined"
        return await self.workflows_service.lint_cron_workflow(
            LintCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
            namespace=self.namespace,
        )

    def build(self) -> TWorkflow:
        """Builds the CronWorkflow and its components into an Argo schema CronWorkflow object."""
        self = self._dispatch_hooks()

        model_workflow = cast(_ModelWorkflow, super().build())
        model_cron_workflow = _ModelCronWorkflow(
            metadata=model_workflow.metadata,
            spec=CronWorkflowSpec(
                workflow_spec=model_workflow.spec,
            ),
        )

        return _CronWorkflowModelMapper.build_model(CronWorkflow, self, model_cron_workflow)

    @classmethod
    def _from_model(cls, model: BaseModel) -> ModelMapperMixin:
        """Parse from given model to cls's type."""
        assert isinstance(model, _ModelCronWorkflow)
        hera_cron_workflow = cls()

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

                if mappers[0].model_path:
                    value = None

                    if (
                        isinstance(mappers[0], _CronWorkflowModelMapper)
                        or isinstance(mappers[0], _WorkflowModelMapper)
                        and mappers[0].model_path[0] == "metadata"
                    ):
                        value = _get_model_attr(model, mappers[0].model_path)

                    elif isinstance(mappers[0], _WorkflowModelMapper) and mappers[0].model_path[0] == "spec":
                        # We map "spec.workflow_spec" from the model CronWorkflow to "spec" for Hera's Workflow (used
                        # as the parent class of Hera's CronWorkflow)
                        value = _get_model_attr(model.spec.workflow_spec, mappers[0].model_path[1:])

                    if value is not None:
                        setattr(hera_cron_workflow, attr, value)

        return hera_cron_workflow

    @classmethod
    def from_dict(cls, model_dict: Dict) -> ModelMapperMixin:
        """Create a CronWorkflow from a CronWorkflow contained in a dict.

        Examples:
            >>> my_cron_workflow = CronWorkflow(name="my-cron-wf")
            >>> my_cron_workflow == CronWorkflow.from_dict(my_cron_workflow.to_dict())
            True
        """
        return cls._from_dict(model_dict, _ModelCronWorkflow)

    @classmethod
    def from_yaml(cls, yaml_str: str) -> ModelMapperMixin:
        """Create a CronWorkflow from a CronWorkflow contained in a YAML string.

        Examples:
            >>> my_cron_workflow = CronWorkflow.from_yaml(yaml_str)
        """
        return cls._from_yaml(yaml_str, _ModelCronWorkflow)

    @classmethod
    def from_file(cls, yaml_file: Union[Path, str]) -> ModelMapperMixin:
        """Create a CronWorkflow from a CronWorkflow contained in a YAML file.

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

    def get_workflow_link(self) -> str:
        """Returns the workflow link for the workflow."""
        assert self.workflows_service is not None, "Cannot fetch a cron workflow link without a service"
        assert self.name is not None, "Cannot fetch a cron workflow link without a cron workflow name"
        return self.workflows_service.get_cron_workflow_link(self.name)

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 | WorkflowLevelArtifactGC
] = None

artifact_repository_ref 

artifact_repository_ref: Optional[ArtifactRepositoryRef] = (
    None
)

automount_service_account_token 

automount_service_account_token: Optional[bool] = None

concurrency_policy 

concurrency_policy: Optional[str] = None

creation_timestamp 

creation_timestamp: Optional[Time] = None

cron_status 

cron_status: Optional[CronWorkflowStatus] = None

cron_suspend 

cron_suspend: Optional[bool] = 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

failed_jobs_history_limit 

failed_jobs_history_limit: Optional[int] = 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

DEPRECATED: The spec.podPriority field was removed in 3.7, so does not map to anything. Use pod_priority_class_name instead.

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[
    Union[RetryStrategy, RetryStrategy]
] = None

schedule 

schedule: Optional[str] = None

scheduler_name 

scheduler_name: Optional[str] = None

schedules 

schedules: Optional[List[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

starting_deadline_seconds 

starting_deadline_seconds: Optional[int] = None

status 

status: Optional[WorkflowStatus] = None

stop_strategy 

stop_strategy: Optional[StopStrategy] = None

successful_jobs_history_limit 

successful_jobs_history_limit: Optional[int] = None

suspend 

suspend: Optional[bool] = None

synchronization 

synchronization: Optional[Synchronization] = None

template_defaults 

template_defaults: Optional[Template] = None

templates 

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

timezone 

timezone: Optional[str] = None

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

when 

when: Optional[str] = None

workflow_metadata 

workflow_metadata: Optional[WorkflowMetadata] = None

workflow_template_ref 

workflow_template_ref: Optional[WorkflowTemplateRef] = None

workflows_service 

workflows_service: Optional[
    Union[WorkflowsService, AsyncWorkflowsService]
] = 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="argoproj/argosay:v2", 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
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="argoproj/argosay:v2", 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_class_name=self.priority_class_name,
            retry_strategy=self._build_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

DEPRECATED: priority has been removed from ModelTemplate.

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[
    Union[RetryStrategy, 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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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 
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
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 isinstance(node, Templatable):
            from hera.workflows.workflow import Workflow

            # We must be under a workflow context due to checks in _HeraContext.add_sub_node
            assert _context.pieces and isinstance(_context.pieces[0], Workflow)
            _context.pieces[0]._add_sub(node)
            return

        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_class_name=self.priority_class_name,
            retry_strategy=self._build_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

DEPRECATED: priority has been removed from ModelTemplate.

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[
    Union[RetryStrategy, 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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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.")

Data

Data implements the Argo data template representation.

Data can be used to indicate that some data, identified by a source, should be processed via the specified transformations. The transformations field can be either expressed via a pure str or via a hera.expr, which transpiles the expression into a statement that can be processed by Argo.

Source code in src/hera/workflows/data.py 
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
class Data(TemplateMixin, IOMixin, CallableTemplateMixin):
    """`Data` implements the Argo data template representation.

    Data can be used to indicate that some data, identified by a `source`, should be processed via the specified
    `transformations`. The `transformations` field can be either expressed via a pure `str` or via a `hera.expr`,
    which transpiles the expression into a statement that can be processed by Argo.
    """

    source: Union[m.DataSource, m.ArtifactPaths, Artifact]
    transformations: List[Union[str, Node]] = []

    def _build_source(self) -> m.DataSource:
        """Builds the generated `DataSource`."""
        if isinstance(self.source, m.DataSource):
            return self.source
        elif isinstance(self.source, m.ArtifactPaths):
            return m.DataSource(artifact_paths=self.source)
        return m.DataSource(artifact_paths=self.source._build_artifact_paths())

    def _build_data(self) -> m.Data:
        """Builds the generated `Data` template."""
        return m.Data(
            source=self._build_source(),
            transformation=list(map(lambda expr: m.TransformationStep(expression=str(expr)), self.transformations)),
        )

    def _build_template(self) -> m.Template:
        """Builds the generated `Template` from the fields of `Data`."""
        return m.Template(
            active_deadline_seconds=self.active_deadline_seconds,
            affinity=self.affinity,
            archive_location=self.archive_location,
            automount_service_account_token=self.automount_service_account_token,
            data=self._build_data(),
            executor=self.executor,
            fail_fast=self.fail_fast,
            host_aliases=self.host_aliases,
            init_containers=self._build_init_containers(),
            inputs=self._build_inputs(),
            outputs=self._build_outputs(),
            memoize=self.memoize,
            metadata=self._build_metadata(),
            name=self.name,
            node_selector=self.node_selector,
            plugin=self.plugin,
            priority_class_name=self.priority_class_name,
            retry_strategy=self._build_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

DEPRECATED: priority has been removed from ModelTemplate.

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[
    Union[RetryStrategy, 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

source 

source: Union[DataSource, ArtifactPaths, Artifact]

synchronization 

synchronization: Optional[Synchronization] = None

timeout 

timeout: Optional[str] = None

tolerations 

tolerations: Optional[List[Toleration]] = None

transformations 

transformations: List[Union[str, Node]] = []

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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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.")

HTTP

HTTP is an implementation of the HTTP template that supports executing HTTP actions in a step/task.

Source code in src/hera/workflows/http_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
class HTTP(TemplateMixin, IOMixin, CallableTemplateMixin):
    """`HTTP` is an implementation of the HTTP template that supports executing HTTP actions in a step/task."""

    url: str
    body: Optional[str] = None
    body_from: Optional[HTTPBodySource] = None
    headers: Optional[List[HTTPHeader]] = None
    insecure_skip_verify: Optional[bool] = None
    method: Optional[str] = None
    success_condition: Optional[str] = None
    timeout_seconds: Optional[int] = None

    def _build_http_template(self) -> _ModelHTTP:
        """Builds the generated HTTP sub-template."""
        return _ModelHTTP(
            url=self.url,
            body=self.body,
            body_from=self.body_from,
            headers=self.headers,
            insecure_skip_verify=self.insecure_skip_verify,
            method=self.method,
            success_condition=self.success_condition,
            timeout_seconds=self.timeout_seconds,
        )

    def _build_template(self) -> _ModelTemplate:
        """Builds the HTTP generated `Template`."""
        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,
            executor=self.executor,
            fail_fast=self.fail_fast,
            host_aliases=self.host_aliases,
            http=self._build_http_template(),
            init_containers=self._build_init_containers(),
            memoize=self.memoize,
            metadata=self._build_metadata(),
            inputs=self._build_inputs(),
            outputs=self._build_outputs(),
            name=self.name,
            node_selector=self.node_selector,
            plugin=self.plugin,
            priority_class_name=self.priority_class_name,
            retry_strategy=self._build_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

body 

body: Optional[str] = None

body_from 

body_from: Optional[HTTPBodySource] = None

daemon 

daemon: Optional[bool] = None

executor 

executor: Optional[ExecutorConfig] = None

fail_fast 

fail_fast: Optional[bool] = None

headers 

headers: Optional[List[V1alpha1HTTPHeader]] = 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

insecure_skip_verify 

insecure_skip_verify: Optional[bool] = None

labels 

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

memoize 

memoize: Optional[Memoize] = None

method 

method: Optional[str] = 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

DEPRECATED: priority has been removed from ModelTemplate.

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[
    Union[RetryStrategy, 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

success_condition 

success_condition: Optional[str] = None

synchronization 

synchronization: Optional[Synchronization] = None

timeout 

timeout: Optional[str] = None

timeout_seconds 

timeout_seconds: Optional[int] = None

tolerations 

tolerations: Optional[List[Toleration]] = None

url 

url: str

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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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 
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
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:
        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 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
@classmethod
def build_model(
    cls, hera_class: Type[ModelMapperMixin], hera_obj: ModelMapperMixin, model: TWorkflow
) -> TWorkflow:
    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

Resource

Resource is a representation of a K8s resource that can be created by Argo.

The resource is a callable step that can be invoked in a DAG/Workflow. The resource can create any K8s resource, such as other workflows, workflow templates, daemons, etc, as specified by the manifest field of the resource. The manifest field is a canonical YAML that is submitted to K8s by Argo. Note that the manifest is a union of multiple types. The manifest can be a string, in which case it is assume to be YAML. Otherwise, if it’s a Hera objects, it is automatically converted to the corresponding YAML representation.

Source code in src/hera/workflows/resource.py 
 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
 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
class Resource(CallableTemplateMixin, TemplateMixin, SubNodeMixin, IOMixin):
    """`Resource` is a representation of a K8s resource that can be created by Argo.

    The resource is a callable step that can be invoked in a DAG/Workflow. The resource can create any K8s resource,
    such as other workflows, workflow templates, daemons, etc, as specified by the `manifest` field of the resource.
    The manifest field is a canonical YAML that is submitted to K8s by Argo. Note that the manifest is a union of
    multiple types. The manifest can be a string, in which case it is assume to be YAML. Otherwise, if it's a Hera
    objects, it is automatically converted to the corresponding YAML representation.
    """

    action: str
    failure_condition: Optional[str] = None
    flags: Optional[List[str]] = None
    manifest: Optional[Union[str, "Workflow", "CronWorkflow", "WorkflowTemplate"]] = None
    manifest_from: Optional[ManifestFrom] = None
    merge_strategy: Optional[str] = None
    set_owner_reference: Optional[bool] = None
    success_condition: Optional[str] = None

    def __new__(cls, *args, **kwargs) -> "Resource":
        """Importing here to avoid circular imports."""
        from hera.workflows.cron_workflow import CronWorkflow
        from hera.workflows.workflow import Workflow
        from hera.workflows.workflow_template import WorkflowTemplate

        cls.update_forward_refs(Workflow=Workflow, CronWorkflow=CronWorkflow, WorkflowTemplate=WorkflowTemplate)

        instance = super().__new__(cls)
        return instance

    def _build_manifest(self) -> Optional[str]:
        from hera.workflows.cron_workflow import CronWorkflow
        from hera.workflows.workflow import Workflow
        from hera.workflows.workflow_template import WorkflowTemplate

        if isinstance(self.manifest, (Workflow, CronWorkflow, WorkflowTemplate)):
            # hack to appease raw yaml string comparison
            return self.manifest.to_yaml().replace("'{{", "{{").replace("}}'", "}}")
        return self.manifest

    def _build_resource_template(self) -> _ModelResourceTemplate:
        return _ModelResourceTemplate(
            action=self.action,
            failure_condition=self.failure_condition,
            flags=self.flags,
            manifest=self._build_manifest(),
            manifest_from=self.manifest_from,
            merge_strategy=self.merge_strategy,
            set_owner_reference=self.set_owner_reference,
            success_condition=self.success_condition,
        )

    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,
            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_class_name=self.priority_class_name,
            resource=self._build_resource_template(),
            retry_strategy=self._build_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,
        )

action 

action: str

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

failure_condition 

failure_condition: Optional[str] = None

flags 

flags: Optional[List[str]] = 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

manifest 

manifest: Optional[
    Union[str, Workflow, CronWorkflow, WorkflowTemplate]
] = None

manifest_from 

manifest_from: Optional[ManifestFrom] = None

memoize 

memoize: Optional[Memoize] = None

merge_strategy 

merge_strategy: Optional[str] = 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

DEPRECATED: priority has been removed from ModelTemplate.

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[
    Union[RetryStrategy, RetryStrategy]
] = None

scheduler_name 

scheduler_name: Optional[str] = None

service_account_name 

service_account_name: Optional[str] = None

set_owner_reference 

set_owner_reference: Optional[bool] = None

sidecars 

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

success_condition 

success_condition: Optional[str] = None

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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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.")

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 
130
131
132
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
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
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_class_name=self.priority_class_name,
                retry_strategy=self._build_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

DEPRECATED: priority has been removed from ModelTemplate.

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[
    Union[RetryStrategy, 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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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 
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
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: Set[str] = 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 isinstance(node, Templatable):
            from hera.workflows.workflow import Workflow

            # We must be under a workflow context due to checks in _HeraContext.add_sub_node
            assert _context.pieces and isinstance(_context.pieces[0], Workflow)
            _context.pieces[0]._add_sub(node)
            return

        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_class_name=self.priority_class_name,
            resource=None,
            retry_strategy=self._build_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

DEPRECATED: priority has been removed from ModelTemplate.

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[
    Union[RetryStrategy, 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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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 
209
210
211
def parallel(self) -> Parallel:
    """Returns a Parallel object which can be used in a sub-context manager."""
    return Parallel()

Suspend

The Suspend template allows the user to pause a workflow for a specified length of time.

The workflow can pause based on the given by duration or indefinitely (i.e. until manually resumed). The Suspend template also allows you to specify intermediate_parameters which will replicate the given parameters to the “inputs” and “outputs” of the template, resulting in a Suspend template that pauses and waits for values from the user for the given list of parameters.

Source code in src/hera/workflows/suspend.py 
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class Suspend(
    TemplateMixin,
    CallableTemplateMixin,
    IOMixin,
):
    """The Suspend template allows the user to pause a workflow for a specified length of time.

    The workflow can pause based on the given by `duration` or indefinitely (i.e. until manually resumed).
    The Suspend template also allows you to specify `intermediate_parameters` which will replicate the given
    parameters to the "inputs" and "outputs" of the template, resulting in a Suspend template that pauses and
    waits for values from the user for the given list of parameters.
    """

    duration: Optional[Union[int, str]] = None
    intermediate_parameters: List[Parameter] = []

    def _build_suspend_template(self) -> _ModelSuspendTemplate:
        return _ModelSuspendTemplate(
            duration=str(self.duration) if self.duration else None,
        )

    def _build_outputs(self) -> Optional[Outputs]:
        intermediate_param_outputs = []
        for param in self.intermediate_parameters:
            intermediate_param_outputs.append(
                Parameter(
                    name=param.name,
                    value_from=ValueFrom(supplied=SuppliedValueFrom()),
                    description=param.description,
                ).as_output()
            )

        all_outputs = super()._build_outputs()
        if all_outputs is not None:
            if all_outputs.parameters is None:
                all_outputs.parameters = []
            all_outputs.parameters.extend(intermediate_param_outputs)
            return all_outputs
        else:
            return Outputs(parameters=intermediate_param_outputs) if intermediate_param_outputs else None

    def _build_inputs(self) -> Optional[Inputs]:
        intermediate_param_inputs = []
        for param in self.intermediate_parameters:
            intermediate_param_inputs.append(param.as_input())

        all_inputs = super()._build_inputs()
        if all_inputs is not None:
            if all_inputs.parameters is None:
                all_inputs.parameters = []
            all_inputs.parameters.extend(intermediate_param_inputs)
            return all_inputs
        else:
            return Inputs(parameters=intermediate_param_inputs) if intermediate_param_inputs else None

    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,
            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(),
            name=self.name,
            node_selector=self.node_selector,
            outputs=self._build_outputs(),
            plugin=self.plugin,
            priority_class_name=self.priority_class_name,
            retry_strategy=self._build_retry_strategy(),
            scheduler_name=self.scheduler_name,
            security_context=self.pod_security_context,
            service_account_name=self.service_account_name,
            sidecars=self._build_sidecars(),
            suspend=self._build_suspend_template(),
            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

duration 

duration: Optional[Union[int, str]] = 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

intermediate_parameters 

intermediate_parameters: List[Parameter] = []

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

DEPRECATED: priority has been removed from ModelTemplate.

priority_class_name 

priority_class_name: Optional[str] = None

retry_strategy 

retry_strategy: Optional[
    Union[RetryStrategy, 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

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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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 
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
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.")

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 
623
624
625
626
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)

async_create 

async_create() -> TWorkflow

Creates the CronWorkflow on the Argo cluster.

Source code in src/hera/workflows/cron_workflow.py 
150
151
152
153
154
155
156
157
158
159
160
161
async def async_create(self) -> TWorkflow:  # type: ignore
    """Creates the CronWorkflow on the Argo cluster."""
    assert isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"

    wf = await self.workflows_service.create_cron_workflow(
        CreateCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
        namespace=self.namespace,
    )
    # set the name on the object so that we can do a get/update later
    self.name = wf.metadata.name
    return wf

async_get 

async_get() -> TWorkflow

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

Source code in src/hera/workflows/cron_workflow.py 
163
164
165
166
167
168
async def async_get(self) -> TWorkflow:
    """Attempts to get a cron workflow based on the parameters of this template e.g. name + namespace."""
    assert isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"
    assert self.name, "workflow name not defined"
    return await self.workflows_service.get_cron_workflow(name=self.name, namespace=self.namespace)

async_lint 

async_lint() -> TWorkflow

Lints the CronWorkflow using the Argo cluster.

Source code in src/hera/workflows/cron_workflow.py 
195
196
197
198
199
200
201
202
async def async_lint(self) -> TWorkflow:
    """Lints the CronWorkflow using the Argo cluster."""
    assert isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"
    return await self.workflows_service.lint_cron_workflow(
        LintCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
        namespace=self.namespace,
    )

async_update 

async_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/cron_workflow.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def async_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 isinstance(self.workflows_service, AsyncWorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"
    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 = await self.async_get()
        template.metadata.resource_version = curr.metadata.resource_version
    except NotFound:
        return await self.async_create()
    return await self.workflows_service.update_cron_workflow(
        self.name,
        UpdateCronWorkflowRequest(cron_workflow=template),  # type: ignore
        namespace=self.namespace,
    )

async_wait 

async_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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
async def async_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 isinstance(self.workflows_service, AsyncWorkflowsService), "workflows 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
    await asyncio.sleep(poll_interval)
    wf = await 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:
        await asyncio.sleep(poll_interval)
        wf = await self.workflows_service.get_workflow(wf.metadata.name, namespace=self.namespace)
        assert wf.metadata.name is not None
        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

build 

build() -> TWorkflow

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

Source code in src/hera/workflows/cron_workflow.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
def build(self) -> TWorkflow:
    """Builds the CronWorkflow and its components into an Argo schema CronWorkflow object."""
    self = self._dispatch_hooks()

    model_workflow = cast(_ModelWorkflow, super().build())
    model_cron_workflow = _ModelCronWorkflow(
        metadata=model_workflow.metadata,
        spec=CronWorkflowSpec(
            workflow_spec=model_workflow.spec,
        ),
    )

    return _CronWorkflowModelMapper.build_model(CronWorkflow, self, model_cron_workflow)

container 

container(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
827
828
829
830
@_add_type_hints(Container)
def container(self, **container_kwargs) -> Callable:
    warnings.warn("`container` is deprecated; use `container_template`.", DeprecationWarning)
    return self.container_template(**container_kwargs)

container_template 

container_template(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
832
833
834
835
836
837
@_add_type_hints(Container)
def container_template(self, **container_kwargs) -> Callable:
    self._check_if_enabled("container")
    from hera.workflows.container import Container

    return self._template_call_decorator_impl(Container, container_kwargs)

create 

create() -> TWorkflow

Creates the CronWorkflow on the Argo cluster.

Source code in src/hera/workflows/cron_workflow.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def create(self) -> TWorkflow:  # type: ignore
    """Creates the CronWorkflow on the Argo cluster."""
    assert isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"

    wf = self.workflows_service.create_cron_workflow(
        CreateCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
        namespace=self.namespace,
    )
    # set the name on the object so that we can do a get/update later
    self.name = wf.metadata.name
    return wf

dag 

dag(**dag_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
1007
1008
1009
1010
1011
1012
1013
@_add_type_hints(DAG)
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)

data_template 

data_template(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
839
840
841
842
843
844
@_add_type_hints(Data)
def data_template(self, **container_kwargs) -> Callable:
    self._check_if_enabled("data")
    from hera.workflows.data import Data

    return self._template_call_decorator_impl(Data, container_kwargs)

from_dict 

from_dict(model_dict: Dict) -> ModelMapperMixin

Create a CronWorkflow from a CronWorkflow contained in a dict.

Examples:

>>> my_cron_workflow = CronWorkflow(name="my-cron-wf")
>>> my_cron_workflow == CronWorkflow.from_dict(my_cron_workflow.to_dict())
True
Source code in src/hera/workflows/cron_workflow.py 
249
250
251
252
253
254
255
256
257
258
@classmethod
def from_dict(cls, model_dict: Dict) -> ModelMapperMixin:
    """Create a CronWorkflow from a CronWorkflow contained in a dict.

    Examples:
        >>> my_cron_workflow = CronWorkflow(name="my-cron-wf")
        >>> my_cron_workflow == CronWorkflow.from_dict(my_cron_workflow.to_dict())
        True
    """
    return cls._from_dict(model_dict, _ModelCronWorkflow)

from_file 

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

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

Examples:

>>> yaml_file = Path(...)
>>> my_workflow_template = CronWorkflow.from_file(yaml_file)
Source code in src/hera/workflows/cron_workflow.py 
269
270
271
272
273
274
275
276
277
@classmethod
def from_file(cls, yaml_file: Union[Path, str]) -> ModelMapperMixin:
    """Create a CronWorkflow from a CronWorkflow contained in a YAML file.

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

from_yaml 

from_yaml(yaml_str: str) -> ModelMapperMixin

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

Examples:

>>> my_cron_workflow = CronWorkflow.from_yaml(yaml_str)
Source code in src/hera/workflows/cron_workflow.py 
260
261
262
263
264
265
266
267
@classmethod
def from_yaml(cls, yaml_str: str) -> ModelMapperMixin:
    """Create a CronWorkflow from a CronWorkflow contained in a YAML string.

    Examples:
        >>> my_cron_workflow = CronWorkflow.from_yaml(yaml_str)
    """
    return cls._from_yaml(yaml_str, _ModelCronWorkflow)

get 

get() -> TWorkflow

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

Source code in src/hera/workflows/cron_workflow.py 
109
110
111
112
113
114
def get(self) -> TWorkflow:
    """Attempts to get a cron workflow based on the parameters of this template e.g. name + namespace."""
    assert isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"
    assert self.name, "workflow name not defined"
    return self.workflows_service.get_cron_workflow(name=self.name, namespace=self.namespace)

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 
331
332
333
334
335
336
337
338
339
340
341
342
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/cron_workflow.py 
279
280
281
282
283
def get_workflow_link(self) -> str:
    """Returns the workflow link for the workflow."""
    assert self.workflows_service is not None, "Cannot fetch a cron workflow link without a service"
    assert self.name is not None, "Cannot fetch a cron workflow link without a cron workflow name"
    return self.workflows_service.get_cron_workflow_link(self.name)

http_template 

http_template(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
846
847
848
849
850
851
@_add_type_hints(HTTP)
def http_template(self, **container_kwargs) -> Callable:
    self._check_if_enabled("http")
    from hera.workflows.http_template import HTTP

    return self._template_call_decorator_impl(HTTP, container_kwargs)

lint 

lint() -> TWorkflow

Lints the CronWorkflow using the Argo cluster.

Source code in src/hera/workflows/cron_workflow.py 
141
142
143
144
145
146
147
148
def lint(self) -> TWorkflow:
    """Lints the CronWorkflow using the Argo cluster."""
    assert isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"
    return self.workflows_service.lint_cron_workflow(
        LintCronWorkflowRequest(cron_workflow=self.build()),  # type: ignore
        namespace=self.namespace,
    )

resource_template 

resource_template(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
853
854
855
856
857
858
@_add_type_hints(Resource)
def resource_template(self, **container_kwargs) -> Callable:
    self._check_if_enabled("resource")
    from hera.workflows.resource import Resource

    return self._template_call_decorator_impl(Resource, container_kwargs)

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 
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
760
761
762
763
764
@_add_type_hints(Script)
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 
609
610
611
612
613
614
615
616
617
618
619
620
621
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 
1015
1016
1017
1018
1019
1020
1021
@_add_type_hints(Steps)
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)

suspend_template 

suspend_template(**container_kwargs) -> Callable
Source code in src/hera/workflows/_meta_mixins.py 
860
861
862
863
864
865
866
@_add_type_hints(Suspend)
def suspend_template(self, **container_kwargs) -> Callable:
    # NOTE: Due to `suspend` being a attribute on the child class, `suspend_template` is used as the decorator name.
    self._check_if_enabled("suspend")
    from hera.workflows.suspend import Suspend

    return self._template_call_decorator_impl(Suspend, container_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 
354
355
356
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 
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
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 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
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/cron_workflow.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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 isinstance(self.workflows_service, WorkflowsService), "workflows service not initialized"
    assert self.namespace, "workflow namespace not defined"
    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_cron_workflow(
        self.name,
        UpdateCronWorkflowRequest(cron_workflow=template),  # type: ignore
        namespace=self.namespace,
    )

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 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
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 isinstance(self.workflows_service, WorkflowsService), "workflows 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.metadata.name is not None
        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