hera#

Submodules#

hera.affinity module#

hera.archive module#

class hera.archive.Archive(disable_compression: bool | None = None, tar_compression_level: int | None = None, zip: bool | None = None)[source]#

Bases: object

Archive sets the archival behavior of artifacts.

disable_compression#

Whether to disable compression intentionally.

Type:

Optional[bool] = None

tar_compression_level#

Specifies the gzip compression level to use for an artifact. The default is the gzip flate.DefaultCompression which is set to -1.

Type:

Optional[int]

zip#

Whether to unzip zipped input artifacts.

Type:

Optional[bool] = None

build() → IoArgoprojWorkflowV1alpha1ArchiveStrategy[source]#
disable_compression: bool | None = None#
tar_compression_level: int | None = None#
zip: bool | None = None#

hera.artifact module#

class hera.artifact.Artifact(name: str, path: str, from_task: str | None = None, sub_path: str | None = None, archive: Archive | None = None)[source]#

Bases: object

An artifact represents an object that Argo creates for a specific task’s output.

The output of a task payload can store specific results at a path, which can then be consumed by another dependent task.

name#

The name of the artifact.

Type:

str

path#

The path to the file to be assembled into an artifact. This path is relative to the task environment. If clients have a volume where results are stored, it’s the path to the created object on the respective volume.

Type:

str

from_task#

The name of the task that generates the artifact.

Type:

Optional[str] = None

archive#

The archive to use for the artifact.

Type:

Optional[Archive] = None

as_argument() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact specifications for use as an argument to a task

as_input() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact specifications for use as an input to a task

as_name(name: str)[source]#

Changes the name of the artifact.

as_output() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact specifications for use as an output of a task

property contains_item: bool#

Check whether the artifact contains an argo item reference

to_path(path: str, sub_path: str | None = None)[source]#

Change the paths of the artifact

class hera.artifact.BucketArtifact(name: str, path: str, bucket: str, key: str, archive: Archive | None = None)[source]#

Bases: Artifact

This artifact is used to represent a bucket object as an artifact.

name#

The name of the input artifact.

Type:

str

path#

The path where to store the input artifact. Note that this path is isolated from the output artifact path of the previous task artifact.

Type:

str

bucket#

The name of the bucket to fetch the artifact from.

Type:

str

key#

Key of the artifact in the bucket.

Type:

str

Notes

Don’t use this directly. Use S3InputArtifact or GCSInputArtifact.

class hera.artifact.GCSArtifact(name: str, path: str, bucket: str, key: str, archive: Archive | None = None)[source]#

Bases: BucketArtifact

GCS artifact specification. See hera.artifact.BucketArtifact

as_argument() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact representation for use as an argument to a task

as_input() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact representation for use as an input to a task

as_output() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact representation for use as an input to a task

class hera.artifact.GitArtifact(name: str, path: str, repo: str, revision: str | None = None, depth: int | None = None, disable_submodules: bool | None = None, fetch: List[str] | None = None, insecure_ignore_host_key: bool | None = None, username_secret_name: str | None = None, username_secret_key: str | None = None, password_secret_name: str | None = None, password_secret_key: str | None = None, ssh_private_key_secret_name: str | None = None, ssh_private_key_secret_key: str | None = None)[source]#

Bases: Artifact

Git artifact representation for fetching Git repositories/artifacts as Argo artifacts.

Parameters:

  • name (str) – Name of the git artifact.

  • path (str) – Path to the artifact.

  • repo (str) – Name of the repo origin.

  • revision (Optional[str] = None) – The revision to fetch e.g. main.

  • depth (Optional[int] = None) – The number of commit to fetch during the clone.

  • disable_submodules (Optional[bool] = None) – Whether to disable git submodules.

  • fetch (Optional[List[str]] = None) – Fetch specifies a number of refs that should be fetched before checkout

  • insecure_ignore_host_key (Optional[bool] = None) – Whether to disable SSH strict host key checking during git clone

  • username_secret_name (Optional[str] = None,) – The secret name to use to fetch the username.

  • username_secret_key (Optional[str] = None) – The key within the username secret to use to fetch the username.

  • password_secret_name (Optional[str] = None,) – The secret name to use to fetch the password.

  • password_secret_key (Optional[str] = None) – The key within the password secret to use to fetch the password.

  • ssh_private_key_secret_name (Optional[str] = None) – The secret name to use to fetch the SSH private key.

  • ssh_private_key_secret_key (Optional[str] = None) – The key within the SSH secret to use to fetch the password.

as_argument() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact for use as an argument of a task

as_input() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact for use as an input to a task

class hera.artifact.HttpArtifact(name: str, path: str, url: str)[source]#

Bases: Artifact

Representation of an HTTP artifact from an arbitrary origin.

Parameters:

  • name (str) – Name to assign to the parameter.

  • path (str) – Path for storing the parameter content.

  • url (str) – URL to fetch the artifact content from.

as_argument() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact for use as an argument of a task

as_input() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact for use as an input to a task

class hera.artifact.RawArtifact(name: str, path: str, data: str)[source]#

Bases: Artifact

This artifact allows raw string content to be placed as an artifact in a container.

Parameters:

  • name (str) – The name of the artifact.

  • path (str) – The container path to the artifact.

  • data (str) – The string contents of the artifact.

as_argument() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact for use as an argument of a task

as_input() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact for use as an input to a task

class hera.artifact.S3Artifact(name: str, path: str, bucket: str, key: str, archive: Archive | None = None)[source]#

Bases: BucketArtifact

S3 artifact specification. See hera.artifact.BucketArtifact

as_argument() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact representation for use as an argument to a task

as_input() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact representation for use as an input to a task

as_output() → IoArgoprojWorkflowV1alpha1Artifact[source]#

Assembles the artifact representation for use as an input to a task

hera.backoff module#

hera.client module#

hera.config module#

hera.cron_workflow module#

hera.dag module#

The implementation of a Hera workflow for Argo-based workflows

class hera.dag.DAG(name: str, inputs: List[Parameter | Artifact] | List[Parameter | Artifact | Dict[str, Any]] | Dict[str, Any] | None = None, outputs: List[Parameter | Artifact] | None = None)[source]#

Bases: IO

A directed acyclic graph representation (workflow) representation.

The DAG/workflow is used as a functional representation for a collection of tasks and steps. The workflow context controls the overall behaviour of tasks, such as whether to notify completion, whether to execute retires, overall parallelism, etc. The workflow can be constructed and submitted to multiple Argo endpoints as long as a token can be associated with the endpoint at the given domain.

Parameters:

  • name (str) – The workflow name. Note that the workflow initiation will replace underscores with dashes.

  • inputs (Optional[) –

    Union[

    List[Union[Parameter, Artifact]], List[Union[Parameter, Artifact, Dict[str, Any]]], Dict[str, Any],

    ]

  • None (] =) – Input or Parameter objects that hold parameter inputs. When a dictionary is specified all the key/value pairs will be transformed into Parameter`s. The `key will be the name field of the Parameter while the value will be the value field of the `Parameter.

:paramInput or Parameter objects that hold parameter inputs. When a dictionary is specified all the key/value

pairs will be transformed into Parameter`s. The `key will be the name field of the Parameter while the value will be the value field of the `Parameter.

Parameters:

outputs (Optional[List[Union[Parameter, Artifact]]] = None) – Any outputs to set on the DAG at a global level.

add_task(t: Task) → DAG[source]#

Add a task to the DAG. Note that tasks are added automatically when the DAG context is used

add_tasks(*ts: Task) → DAG[source]#

Add a collection of tasks to the DAG.

Note that tasks are added automatically when the DAG context is used

build() → List[IoArgoprojWorkflowV1alpha1Template][source]#

Assembles the main DAG/workflow template

get_parameter(name: str) → Parameter[source]#

Returns a DAG output as a Parameter to use an input somewhere else

inputs: List[Parameter | Artifact] | List[Parameter | Artifact | Dict[str, Any]] | Dict[str, Any]#
outputs: List[Parameter | Artifact]#

hera.env module#

class hera.env.ConfigMapEnv(config_map_name: str, config_map_key: str, name: str, value: Any | None = None, value_from_input: str | Parameter | None = None)[source]#

Bases: Env, ConfigMapNamedKey

Environment variable specification from K8S config map.

config_map_name#

The name of the config map to load values from.

Type:

str

config_map_key#

The key of the value within the config map.

Type:

str

build() → EnvVar[source]#

Constructs and returns the Argo environment specification

name: str#
class hera.env.ConfigMapNamedKey(config_map_name: str, config_map_key: str)[source]#

Bases: object

Config map representation. Supports the specification of a name/key string pair to identify a value

config_map_key: str#
config_map_name: str#
class hera.env.Env(name: str, value: Any | None = None, value_from_input: str | Parameter | None = None)[source]#

Bases: object

Environment variable specification for tasks.

name#

The name of the variable.

Type:

str

value#

The value of the variable. This value is serialized for the client. It is up to the client to deserialize the value in the task. In addition, if another type is passed, covered by Any, an attempt at json.dumps will be performed.

Type:

Optional[Any] = None

value_from_input#

An external reference which will resolve the env-value, E.g. another task’s output parameter. An input parameter will be auto-generated for the task the Env is instantiated in.

Type:

Optional[Union[str, Parameter]] = None

Raises:

AssertionError – When the specified value is not JSON serializable.

build() → EnvVar[source]#

Constructs and returns the Argo environment specification

name: str#
property param_name: str#
value: Any | None = None#
value_from_input: str | Parameter | None = None#
class hera.env.FieldEnv(field_path: str, name: str, value: Any | None = None, value_from_input: str | Parameter | None = None, api_version: str | None = 'v1')[source]#

Bases: Env, FieldPath

Environment variable specification from K8S object field.

name#

The name of the variable.

Type:

str

value#

The value of the variable. This value is serialized for the client. It is up to the client to deserialize the value in the task. In addition, if another type is passed, covered by Any, an attempt at json.dumps will be performed.

Type:

Optional[Any] = None

value_from_input#

A reference to an input parameter which will resolve to the value. The input parameter will be auto-generated.

Type:

Optional[str] = None

field_path#

Path to the field to obtain the value from.

Type:

str

api_version#

The version of the schema the FieldPath is written in terms of. Defaults to ‘v1’.

Type:

Optional[str] = ‘v1’

api_version: str | None = 'v1'#
build() → EnvVar[source]#

Constructs and returns the Argo environment specification

class hera.env.FieldPath(field_path: str)[source]#

Bases: object

Field path representation.

This allows obtaining K8S values via indexing into specific fields of the K8S definition.

field_path#

Path to the field to obtain the value from.

Type:

str

field_path: str#
class hera.env.SecretEnv(secret_name: str, secret_key: str, name: str, value: Any | None = None, value_from_input: str | Parameter | None = None)[source]#

Bases: Env, SecretNamedKey

Environment variable specification from K8S secrets.

secret_name#

The name of the secret to load values from.

Type:

str

secret_key#

The key of the value within the secret.

Type:

str

build() → EnvVar[source]#

Constructs and returns the Argo environment specification

name: str#
class hera.env.SecretNamedKey(secret_name: str, secret_key: str)[source]#

Bases: object

Secret map representation. Supports the specification of a name/key string pair to identify a value

secret_key: str#
secret_name: str#

hera.env_from module#

class hera.env_from.BaseEnvFrom(prefix: str = '')[source]#

Bases: object

Environment variable specification from K8S resources.

prefix#

An optional identifier to prepend to each key in the specified resources.

Type:

str = ‘’

build() → EnvFromSource[source]#

Constructs and returns the Argo EnvFrom specification

prefix: str = ''#
class hera.env_from.ConfigMapEnvFrom(config_map_name: str, prefix: str = '', optional: bool = False)[source]#

Bases: BaseEnvFrom, _NamedConfigMap

Environment variable specification from K8S config map.

config_map_name#

The name of the config map to load environments.

Type:

str

optional#

Specify whether the K8S config map must be defined

Type:

bool = False

build() → EnvFromSource[source]#

Constructs and returns the Argo EnvFrom specification

optional: bool = False#
class hera.env_from.SecretEnvFrom(secret_name: str, prefix: str = '', optional: bool = False)[source]#

Bases: BaseEnvFrom, _NamedSecret

Environment variable specification from K8S secrets.

secret_name#

The name of the secret to load environments.

Type:

str

optional#

Specify whether the K8S secret must be defined

Type:

bool = False

build() → EnvFromSource[source]#

Constructs and returns the Argo EnvFrom specification

optional: bool = False#

hera.host_alias module#

hera.host_config module#

hera.image module#

hera.io module#

class hera.io.IO(inputs: ~typing.List[~hera.parameter.Parameter | ~hera.artifact.Artifact] | ~typing.List[~hera.parameter.Parameter | ~hera.artifact.Artifact | ~typing.Dict[str, ~typing.Any]] | ~typing.Dict[str, ~typing.Any] = <factory>, outputs: ~typing.List[~hera.parameter.Parameter | ~hera.artifact.Artifact] = <factory>)[source]#

Bases: object

Input/output high-level representation.

Parameters:

  • inputs (Union[) – List[Union[Parameter, Artifact]], List[Union[Parameter, Artifact, Dict[str, Any]]], Dict[str, Any],

  • None (] =) – Input or Parameter objects that hold parameter inputs. When a dictionary is specified all the key/value pairs will be transformed into Parameter`s. The `key will be the name field of the Parameter while the value will be the value field of the `Parameter.

:paramInput or Parameter objects that hold parameter inputs. When a dictionary is specified all the key/value

pairs will be transformed into Parameter`s. The `key will be the name field of the Parameter while the value will be the value field of the `Parameter.

Parameters:

outputs (List[Union[Parameter, Artifact]]) – List of parameters or artifacts to use as outputs.

inputs: List[Parameter | Artifact] | List[Parameter | Artifact | Dict[str, Any]] | Dict[str, Any]#
outputs: List[Parameter | Artifact]#

hera.memoize module#

hera.metric module#

hera.operator module#

hera.parameter module#

Holds input model specifications

class hera.parameter.Parameter(name: str, value: ~typing.Any | None = <object object>, default: str | None = <object object>, value_from: ~hera.value_from.ValueFrom | None = None, description: str | None = None, enum: ~typing.List[str] | None = None, global_name: str | None = None)[source]#

Bases: object

A representation of input from one task to another.

Parameters:

  • name (str) – The name of the task to take input from. The task’s results are expected via stdout. Specifically, the task is expected to perform the script illustrated in Examples.

  • value (Optional[Any] = None) – Value of the parameter, as an index into some field of the task. If this is left as None, along with value_from being left as None, as is the case in GitOps patterns, the submitter has to likely supply the parameter value via the Argo CLI. Note that if a value is supplied a json.dumps will be applied to it.

  • default (Optional[str] = None) – Default value of the parameter in case the value cannot be obtained based on the specification.

  • value_from (Optional[ValueFrom] = None) – Describes a location in which to obtain the value to a parameter. See hera.value_from.ValueFrom or https://argoproj.github.io/argo-workflows/fields/#valuefrom.

  • description (Optional[str] = None) – An optional parameter description.

  • enum (Optional[List[str]] = None) – Holds a list of string values to choose from, for the actual value of the parameter.

  • global_name (Optional[str] = None) – Exports an output parameter to the global scope, making it available as ‘{{workflow.outputs.parameters.XXXX}} and in workflow.status.outputs.parameters.

as_argument() → IoArgoprojWorkflowV1alpha1Parameter | None[source]#

Assembles the parameter for use as an argument of a task

as_input() → IoArgoprojWorkflowV1alpha1Parameter[source]#

Assembles the parameter for use as an input to task

as_name(name: str) → Parameter[source]#

Changes the name of the parameter.

as_output() → IoArgoprojWorkflowV1alpha1Parameter[source]#

Assembles the parameter for use as an output of a task

property contains_item: bool#

Check whether the parameter contains an argo item reference

hera.resource_template module#

hera.resources module#

hera.retry_policy module#

hera.retry_strategy module#

hera.security_context module#

hera.sequence module#

hera.task module#

hera.template_ref module#

hera.toleration module#

hera.ttl_strategy module#

hera.validators module#

Holds a collection of validators that are shared in V1

hera.validators.validate_name(name: str, max_length: int | None = None, generate_name: bool = False) → str[source]#

Validates a name according to standard argo/kubernetes limitations

Parameters:

  • name (str) – The name which should be validated.

  • max_length (Optional[int] = None) – Specify a maximum length of the name. Example: Kubernetes labels have a maximum length of 63 characters.

  • generate_name (bool = False) – Whether the provided name is to be used as a prefix for name generation. If set, name is allowed to end in a single dot (.) or any number of hyphens (-).

Raises:

ValueError – When the name is invalid according to specifications.

Notes

Official doc on object names in Kubernetes: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/

hera.validators.validate_storage_units(value: str) → None[source]#

Validates the units of the given value.

The given value is expected to satisfy a unit/value format that specifies a resource requirement such as 500Mi, 1Gi, etc.

Parameters:

value (str) – The value to validate the units of.

Raises:

  • ValueError – When the units cannot be extracted from the given value.

  • AssertionError – When the identified unit is not a supported one. The supported units are [‘Ki’, ‘Mi’, ‘Gi’, ‘Ti’, ‘Pi’, ‘Ei’].

hera.value_from module#

class hera.value_from.ConfigMapKeyRef(key: str, name: str, optional: bool = False)[source]#

Bases: object

Holds the ConfigMap reference.

key#

The key to select.

Type:

str

name#

The ConfigMap to select from.

Type:

str

optional#

Specify whether the ConfigMap or its key must be defined.

Type:

bool

key: str#
name: str#
optional: bool = False#
class hera.value_from.ValueFrom(config_map_key_ref: ConfigMapKeyRef | None = None, default: str | None = None, event: str | None = None, expression: str | None = None, jq_filter: str | None = None, json_path: str | None = None, parameter: str | None = None, path: str | None = None, supplied: bool = False)[source]#

Bases: object

Holds descriptions of where to obtain parameter values from.

config_map_key_ref#

A selector that identifies a config map to obtain the value from.

Type:

Optional[ConfigMapKeyRef] = None

default#

Specifies a value to be used if retrieving the value from the specified source fails.

Type:

Optional[str] = None

event#

Selector that is evaluated against the event to get the value of the parameter e.g. ‘payload.message’. See antonmedv/expr for more examples and full documentation as it is the package used by Argo.

Type:

Optional[str] = None

expression#

A specification of the value of the parameter.

Type:

Optional[str] = None

jq_filter#

JQFilter expression against the resource object in resource templates to obtain a value from.

Type:

Optional[str] = None

json_path#

The JSON path of a resource to retrieve an output parameter value from in resource templates.

Type:

Optional[str] = None

parameter#

Parameter reference to a task or DAG in which to retrieve an output parameter value from, e.g. ‘{{tasks.t.outputs.param}}’). Note that you can obtain such specification via task.get_parameter(…), dag.get_parameter(…), and workflow.get_parameter(…).

Type:

Optional[str] = None

path#

Path in the container to retrieve an output parameter value from in container templates.

Type:

Optional[str] = None

supplied#

Whether the value will be filled via CLI, API, UI, etc.

Type:

bool = False

Notes

See: https://argoproj.github.io/argo-workflows/fields/#valuefrom

build() → IoArgoprojWorkflowV1alpha1ValueFrom[source]#
config_map_key_ref: ConfigMapKeyRef | None = None#
default: str | None = None#
event: str | None = None#
expression: str | None = None#
jq_filter: str | None = None#
json_path: str | None = None#
parameter: str | None = None#
path: str | None = None#
supplied: bool = False#

hera.volume_claim_gc module#

hera.volumes module#

hera.workflow module#

hera.workflow_service module#

hera.workflow_status module#

hera.workflow_template module#

Module contents#