Env

The hera.workflows.env module provides implementations of environment variable types that can be used with Argo.

Env classes differ from EnvFrom classes as EnvFrom uses a source to retrieve a variable from, and you can only prefix the name with something. The Env classes can create new independent variables.

Env

A variable implementation that can expose a plain value or value from input as an env variable.

Source code in src/hera/workflows/env.py

@dataclass(kw_only=True)
class Env(_BaseEnv):
    """A variable implementation that can expose a plain `value` or `value from input` as an env variable."""

    value: Optional[Any] = None
    """the value of the environment variable"""

    value_from_input: Optional[Union[str, Parameter]] = None
    """an optional `str` or `Parameter` representation of the origin of the environment variable value"""

    @staticmethod
    def _sanitise_param_for_argo(v: str) -> str:
        """Sanitizes the given `v` value into one that satisfies Argo's parameter rules.

        Argo has some strict parameter validation. To satisfy, we replace all ._ with a dash,
        take only first 32 characters from a-zA-Z0-9-, and append md5 digest of the original string.
        """
        # NOTE move this to some general purpose utils?
        replaced_dashes = v.translate(str.maketrans({e: "-" for e in "_."}))  # type: ignore
        legit_set = string.ascii_letters + string.digits + "-"
        legit_prefix = "".join(islice((c for c in replaced_dashes if c in legit_set), 32))
        hash_suffix = hashlib.md5(v.encode("utf-8")).hexdigest()
        return f"{legit_prefix}-{hash_suffix}"

    def __post_init__(self):
        """Validates that only one of `value` or `value_from_input` is specified."""
        if self.value is not None and self.value_from_input is not None:
            raise ValueError("cannot specify both `value` and `value_from_input`")

    @property
    def param_name(self) -> str:
        """Returns the parameter name of the environment variable, conditioned on the use of `value_from_input`."""
        if not self.value_from_input:
            raise ValueError(
                "Unexpected use of `param_name` - without `value_from_input`, no param should be generated"
            )
        return Env._sanitise_param_for_argo(self.name)

    def build(self) -> _ModelEnvVar:
        """Constructs and returns the Argo environment specification."""
        if self.value_from_input is not None:
            self.value = f"{{{{inputs.parameters.{self.param_name}}}}}"
        elif isinstance(self.value, str):
            self.value = self.value
        else:
            self.value = json.dumps(self.value)
        return _ModelEnvVar(name=self.name, value=self.value)

name

name: str

the name of the environment variable. This is universally required irrespective of the type of env variable

param_name

param_name: str

Returns the parameter name of the environment variable, conditioned on the use of value_from_input.

value

value: Optional[Any] = None

the value of the environment variable

value_from_input

value_from_input: Optional[Union[str, Parameter]] = None

an optional str or Parameter representation of the origin of the environment variable value

build

build() -> EnvVar

Constructs and returns the Argo environment specification.

Source code in src/hera/workflows/env.py

def build(self) -> _ModelEnvVar:
    """Constructs and returns the Argo environment specification."""
    if self.value_from_input is not None:
        self.value = f"{{{{inputs.parameters.{self.param_name}}}}}"
    elif isinstance(self.value, str):
        self.value = self.value
    else:
        self.value = json.dumps(self.value)
    return _ModelEnvVar(name=self.name, value=self.value)

FieldEnv

FieldEnv is an environment variable whose origin is in a field specification.

The field path specification points to a particular field within the workflow/container YAML specification. For instance, if there’s a YAML that has 3 fields like so

name: abc
spec:
  a: 42

then a reference to the field a must be encoded as spec.a in order for the value of 42 to be extracted and set as an environment variable.

Source code in src/hera/workflows/env.py

@dataclass(kw_only=True)
class FieldEnv(_BaseEnv):
    """`FieldEnv` is an environment variable whose origin is in a field specification.

    The field path specification points to a particular field within the workflow/container YAML specification. For
    instance, if there's a YAML that has 3 fields like so
    ```
    name: abc
    spec:
      a: 42
    ```
    then a reference to the field a must be encoded as `spec.a` in order for the value of `42` to be extracted and set
    as an environment variable.
    """

    field_path: str
    """the path to the field whose value should be extracted into an environment variable"""

    api_version: Optional[str] = None
    """optionally, an API version specification."""

    def build(self) -> _ModelEnvVar:
        """Constructs and returns the Argo environment specification."""
        return _ModelEnvVar(
            name=self.name,
            value_from=_ModelEnvVarSource(
                field_ref=_ModelObjectFieldSelector(
                    field_path=self.field_path,
                    api_version=self.api_version,
                )
            ),
        )

api_version

api_version: Optional[str] = None

optionally, an API version specification.

field_path

field_path: str

the path to the field whose value should be extracted into an environment variable

name

name: str

the name of the environment variable. This is universally required irrespective of the type of env variable

build

build() -> EnvVar

Constructs and returns the Argo environment specification.

Source code in src/hera/workflows/env.py

def build(self) -> _ModelEnvVar:
    """Constructs and returns the Argo environment specification."""
    return _ModelEnvVar(
        name=self.name,
        value_from=_ModelEnvVarSource(
            field_ref=_ModelObjectFieldSelector(
                field_path=self.field_path,
                api_version=self.api_version,
            )
        ),
    )

ResourceEnv

ResourceEnv exposes a resource field as an environment variable.

Only resources limits and requests such as limits.cpu, limits.memory, limits.ephemeral-storage, requests.cpu, requests.memory and requests.ephemeral-storage are currently supported.

Source code in src/hera/workflows/env.py

@dataclass(kw_only=True)
class ResourceEnv(_BaseEnv):
    """`ResourceEnv` exposes a resource field as an environment variable.

    Only resources limits and requests such as `limits.cpu`, `limits.memory`, `limits.ephemeral-storage`,
    `requests.cpu`, `requests.memory` and `requests.ephemeral-storage` are currently supported.
    """

    resource: str
    """the name of the resource to select, such as `limit.cpu`, `limits.memory`, etc."""

    container_name: Optional[str] = None
    """
    a pod can contain multiple containers, so this field helps select the right container whose resources should 
    be exposed as an env variable.
    """

    divisor: Optional[Quantity] = None
    """Specifies the output format of the exposed resources, defaults to `1` on Argo's side"""

    def build(self) -> _ModelEnvVar:
        """Builds the `ResourceEnv` into a Hera auto-generated environment variable model."""
        return _ModelEnvVar(
            name=self.name,
            value_from=_ModelEnvVarSource(
                resource_field_ref=_ModelResourceFieldSelector(
                    container_name=self.container_name,
                    divisor=self.divisor,
                    resource=self.resource,
                )
            ),
        )

container_name

container_name: Optional[str] = None

a pod can contain multiple containers, so this field helps select the right container whose resources should be exposed as an env variable.

divisor

divisor: Optional[Quantity] = None

Specifies the output format of the exposed resources, defaults to 1 on Argo’s side

name

name: str

the name of the environment variable. This is universally required irrespective of the type of env variable

resource

resource: str

the name of the resource to select, such as limit.cpu, limits.memory, etc.

build

build() -> EnvVar

Builds the ResourceEnv into a Hera auto-generated environment variable model.

Source code in src/hera/workflows/env.py

def build(self) -> _ModelEnvVar:
    """Builds the `ResourceEnv` into a Hera auto-generated environment variable model."""
    return _ModelEnvVar(
        name=self.name,
        value_from=_ModelEnvVarSource(
            resource_field_ref=_ModelResourceFieldSelector(
                container_name=self.container_name,
                divisor=self.divisor,
                resource=self.resource,
            )
        ),
    )

SecretEnv

SecretEnv is an environment variable whose value originates from a Kubernetes secret.

Source code in src/hera/workflows/env.py

@dataclass(kw_only=True)
class SecretEnv(_BaseEnv):
    """`SecretEnv` is an environment variable whose value originates from a Kubernetes secret."""

    secret_key: str
    """the field key within the secret that points to the value to extract and set as an env variable"""

    secret_name: Optional[str] = None
    """the name of the Kubernetes secret to extract the value from"""

    optional: Optional[bool] = None
    """whether the existence of the secret is optional"""

    def build(self) -> _ModelEnvVar:
        """Constructs and returns the Argo environment specification."""
        return _ModelEnvVar(
            name=self.name,
            value_from=_ModelEnvVarSource(
                secret_key_ref=_ModelSecretKeySelector(
                    name=self.secret_name, key=self.secret_key, optional=self.optional
                )
            ),
        )

name

name: str

the name of the environment variable. This is universally required irrespective of the type of env variable

optional

optional: Optional[bool] = None

whether the existence of the secret is optional

secret_key

secret_key: str

the field key within the secret that points to the value to extract and set as an env variable

secret_name

secret_name: Optional[str] = None

the name of the Kubernetes secret to extract the value from

build

build() -> EnvVar

Constructs and returns the Argo environment specification.

Source code in src/hera/workflows/env.py

def build(self) -> _ModelEnvVar:
    """Constructs and returns the Argo environment specification."""
    return _ModelEnvVar(
        name=self.name,
        value_from=_ModelEnvVarSource(
            secret_key_ref=_ModelSecretKeySelector(
                name=self.secret_name, key=self.secret_key, optional=self.optional
            )
        ),
    )

Comments