First-Party Volumes

The hera.workflows.volume module provides all Argo volume types that can be used via Hera.

EmptyDirVolume

Representation of an empty dir volume from K8s.

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

@dataclass(kw_only=True)
class EmptyDirVolume(_BaseVolume):
    """Representation of an empty dir volume from K8s."""

    medium: Optional[str] = None
    size_limit: Optional[resource.Quantity] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name, empty_dir=_ModelEmptyDirVolumeSource(medium=self.medium, size_limit=self.size_limit)
        )

medium

medium: Optional[str] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

size_limit

size_limit: Optional[Quantity] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

ExistingVolume

ExistingVolume is a representation of an existing volume in K8s.

The existing volume is mounted based on the supplied claim name. This tells K8s that the specified persistent volume claim should be used to mount a volume to a pod.

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

@dataclass(kw_only=True)
class ExistingVolume(_BaseVolume):
    """`ExistingVolume` is a representation of an existing volume in K8s.

    The existing volume is mounted based on the supplied claim name. This tells K8s that the specified persistent
    volume claim should be used to mount a volume to a pod.
    """

    claim_name: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            persistent_volume_claim=_ModelPersistentVolumeClaimVolumeSource(
                claim_name=self.claim_name, read_only=self.read_only
            ),
        )

claim_name

claim_name: str

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

ConfigMapVolume

Representation of a config map volume.

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

@dataclass(kw_only=True)
class ConfigMapVolume(_BaseVolume):
    """Representation of a config map volume."""

    default_mode: Optional[int] = None
    items: Optional[List[KeyToPath]] = None
    optional: Optional[bool] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            config_map=_ModelConfigMapVolumeSource(
                default_mode=self.default_mode, items=self.items, name=self.name, optional=self.optional
            ),
        )

default_mode

default_mode: Optional[int] = None

items

items: Optional[List[KeyToPath]] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

optional

optional: Optional[bool] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

DownwardAPIVolume

Representation of a volume passed via the downward API.

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

@dataclass(kw_only=True)
class DownwardAPIVolume(_BaseVolume):
    """Representation of a volume passed via the downward API."""

    default_mode: Optional[int] = None
    items: Optional[List[DownwardAPIVolumeFile]] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            downward_api=_ModelDownwardAPIVolumeSource(default_mode=self.default_mode, items=self.items),
        )

default_mode

default_mode: Optional[int] = None

items

items: Optional[List[DownwardAPIVolumeFile]] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

EphemeralVolume

Representation of a volume that uses ephemeral storage shared with the K8s node a pod is scheduled on.

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

@dataclass(kw_only=True)
class EphemeralVolume(_BaseVolume):
    """Representation of a volume that uses ephemeral storage shared with the K8s node a pod is scheduled on."""

    volume_claim_template: Optional[PersistentVolumeClaimTemplate] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name, ephemeral=_ModelEphemeralVolumeSource(volume_claim_template=self.volume_claim_template)
        )

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

volume_claim_template

volume_claim_template: Optional[
    PersistentVolumeClaimTemplate
] = None

FlexVolume

A Flex volume representation.

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

@dataclass(kw_only=True)
class FlexVolume(_BaseVolume):
    """A Flex volume representation."""

    driver: str
    fs_type: Optional[str] = None
    options: Optional[Dict[str, str]] = None
    secret_ref: Optional[LocalObjectReference] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            flex_volume=_ModelFlexVolumeSource(
                driver=self.driver,
                fs_type=self.fs_type,
                options=self.options,
                read_only=self.read_only,
                secret_ref=self.secret_ref,
            ),
        )

driver

driver: str

fs_type

fs_type: Optional[str] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

options

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

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

secret_ref

secret_ref: Optional[LocalObjectReference] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

SecretVolume

SecretVolume supports mounting a K8s secret as a container volume.

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

@dataclass(kw_only=True)
class SecretVolume(_BaseVolume):
    """`SecretVolume` supports mounting a K8s secret as a container volume."""

    default_mode: Optional[int] = None
    items: Optional[List[KeyToPath]] = None
    optional: Optional[bool] = None
    secret_name: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            secret=_ModelSecretVolumeSource(
                default_mode=self.default_mode, items=self.items, optional=self.optional, secret_name=self.secret_name
            ),
        )

default_mode

default_mode: Optional[int] = None

items

items: Optional[List[KeyToPath]] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

optional

optional: Optional[bool] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

secret_name

secret_name: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

HostPathVolume

Representation for a volume that can be mounted from a host path/node location.

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

@dataclass(kw_only=True)
class HostPathVolume(_BaseVolume):
    """Representation for a volume that can be mounted from a host path/node location."""

    path: str
    type: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(name=self.name, host_path=_ModelHostPathVolumeSource(path=self.path, type=self.type))

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

path

path: str

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

type

type: Optional[str] = None

NFSVolume

A network file system volume representation.

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

@dataclass(kw_only=True)
class NFSVolume(_BaseVolume):
    """A network file system volume representation."""

    path: str
    server: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name, nfs=_ModelNFSVolumeSource(server=self.server, path=self.path, read_only=self.read_only)
        )

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

path

path: str

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

server

server: str

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

ProjectedVolume

ProjectedVolume represents a projected volume to mount to a container.

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

@dataclass(kw_only=True)
class ProjectedVolume(_BaseVolume):
    """`ProjectedVolume` represents a projected volume to mount to a container."""

    default_mode: Optional[int] = None
    sources: Optional[List[VolumeProjection]] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name, projected=_ModelProjectedVolumeSource(default_mode=self.default_mode, sources=self.sources)
        )

default_mode

default_mode: Optional[int] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

sources

sources: Optional[List[VolumeProjection]] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

Comments