Third-Party Volumes

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

AWSElasticBlockStoreVolume

Representation of AWS elastic block store volume.

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

@dataclass(kw_only=True)
class AWSElasticBlockStoreVolume(_BaseVolume):
    """Representation of AWS elastic block store volume."""

    volume_id: str
    fs_type: Optional[str] = None
    partition: Optional[int] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            aws_elastic_block_store=_ModelAWSElasticBlockStoreVolumeSource(
                fs_type=self.fs_type, partition=self.partition, read_only=self.read_only, volume_id=self.volume_id
            ),
        )

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

partition

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

volume_id: str

AzureDiskVolume

Representation of an Azure disk volume.

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

@dataclass(kw_only=True)
class AzureDiskVolume(_BaseVolume):
    """Representation of an Azure disk volume."""

    caching_mode: Optional[str] = None
    disk_name: str
    disk_uri: str
    fs_type: Optional[str] = None
    kind: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            azure_disk=_ModelAzureDiskVolumeSource(
                caching_mode=self.caching_mode,
                disk_name=self.disk_name,
                disk_uri=self.disk_uri,
                fs_type=self.fs_type,
                kind=self.kind,
                read_only=self.read_only,
            ),
        )

caching_mode

caching_mode: Optional[str] = None

disk_name

disk_name: str

disk_uri

disk_uri: str

fs_type

fs_type: Optional[str] = None

kind

kind: 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

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

AzureFileVolume

Representation of an Azure file that can be mounted as a volume.

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

@dataclass(kw_only=True)
class AzureFileVolume(_BaseVolume):
    """Representation of an Azure file that can be mounted as a volume."""

    secret_name: str
    share_name: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            azure_file=_ModelAzureFileVolumeSource(
                read_only=self.read_only, secret_name=self.secret_name, share_name=self.share_name
            ),
        )

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

secret_name

secret_name: str

share_name

share_name: str

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

CephFSVolume

Representation of a Ceph file system volume.

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

@dataclass(kw_only=True)
class CephFSVolume(_BaseVolume):
    """Representation of a Ceph file system volume."""

    monitors: List[str]
    path: Optional[str] = None
    secret_file: Optional[str] = None
    secret_ref: Optional[LocalObjectReference] = None
    user: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            cephfs=_ModelCephFSVolumeSource(
                monitors=self.monitors,
                path=self.path,
                read_only=self.read_only,
                secret_file=self.secret_file,
                secret_ref=self.secret_ref,
                user=self.user,
            ),
        )

monitors

monitors: List[str]

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

path

path: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

secret_file

secret_file: 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

user

user: Optional[str] = None

CinderVolume

Representation of a Cinder volume.

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

@dataclass(kw_only=True)
class CinderVolume(_BaseVolume):
    """Representation of a Cinder volume."""

    fs_type: Optional[str] = None
    secret_ref: Optional[LocalObjectReference] = None
    volume_id: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            cinder=_ModelCinderVolumeSource(
                fs_type=self.fs_type,
                read_only=self.read_only,
                secret_ref=self.secret_ref,
                volume_id=self.volume_id,
            ),
        )

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

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

volume_id

volume_id: str

CSIVolume

Representation of a container service interface volume.

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

@dataclass(kw_only=True)
class CSIVolume(_BaseVolume):
    """Representation of a container service interface volume."""

    driver: str
    fs_type: Optional[str] = None
    node_publish_secret_ref: Optional[LocalObjectReference] = None
    volume_attributes: Optional[Dict[str, str]] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            csi=_ModelCSIVolumeSource(
                driver=self.driver,
                fs_type=self.fs_type,
                node_publish_secret_ref=self.node_publish_secret_ref,
                read_only=self.read_only,
                volume_attributes=self.volume_attributes,
            ),
        )

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

node_publish_secret_ref

node_publish_secret_ref: Optional[LocalObjectReference] = (
    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_attributes

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

FCVolume

An FV volume representation.

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

@dataclass(kw_only=True)
class FCVolume(_BaseVolume):
    """An FV volume representation."""

    fs_type: Optional[str] = None
    lun: Optional[int] = None
    target_ww_ns: Optional[List[str]] = None
    wwids: Optional[List[str]] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            fc=_ModelFCVolumeSource(
                fs_type=self.fs_type,
                lun=self.lun,
                read_only=self.read_only,
                target_ww_ns=self.target_ww_ns,
                wwids=self.wwids,
            ),
        )

fs_type

fs_type: Optional[str] = None

lun

lun: 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

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

target_ww_ns

target_ww_ns: Optional[List[str]] = None

wwids

wwids: Optional[List[str]] = None

FlockerVolume

A Flocker volume representation.

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

@dataclass(kw_only=True)
class FlockerVolume(_BaseVolume):
    """A Flocker volume representation."""

    dataset_name: Optional[str] = None
    dataset_uuid: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            flocker=_ModelFlockerVolumeSource(dataset_name=self.dataset_name, dataset_uuid=self.dataset_uuid),
        )

dataset_name

dataset_name: Optional[str] = None

dataset_uuid

dataset_uuid: 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

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

GCEPersistentDiskVolume

A representation of a Google Cloud Compute Enginer persistent disk.

Notes

The volume must exist on GCE before a request to mount it to a pod is performed.

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

@dataclass(kw_only=True)
class GCEPersistentDiskVolume(_BaseVolume):
    """A representation of a Google Cloud Compute Enginer persistent disk.

    Notes:
        The volume must exist on GCE before a request to mount it to a pod is performed.
    """

    fs_type: Optional[str] = None
    partition: Optional[int] = None
    pd_name: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            gce_persistent_disk=_ModelGCEPersistentDiskVolumeSource(
                fs_type=self.fs_type, partition=self.partition, pd_name=self.pd_name, read_only=self.read_only
            ),
        )

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

partition

partition: Optional[int] = None

pd_name

pd_name: 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

GitRepoVolume

A representation of a Git repo that can be mounted as a volume.

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

@dataclass(kw_only=True)
class GitRepoVolume(_BaseVolume):
    """A representation of a Git repo that can be mounted as a volume."""

    directory: Optional[str] = None
    repository: str
    revision: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            git_repo=_ModelGitRepoVolumeSource(
                directory=self.directory, repository=self.repository, revision=self.revision
            ),
        )

directory

directory: 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

repository

repository: str

revision

revision: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

GlusterfsVolume

A representation for a Gluster filesystem volume.

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

@dataclass(kw_only=True)
class GlusterfsVolume(_BaseVolume):
    """A representation for a Gluster filesystem volume."""

    endpoints: str
    path: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            glusterfs=_ModelGlusterfsVolumeSource(endpoints=self.endpoints, path=self.path, read_only=self.read_only),
        )

endpoints

endpoints: str

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

ISCSIVolume

Representation of ISCSI volume.

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

@dataclass(kw_only=True)
class ISCSIVolume(_BaseVolume):
    """Representation of ISCSI volume."""

    chap_auth_discovery: Optional[bool] = None
    chap_auth_session: Optional[bool] = None
    fs_type: Optional[str] = None
    initiator_name: Optional[str] = None
    iqn: str
    iscsi_interface: Optional[str] = None
    lun: int
    portals: Optional[List[str]] = None
    secret_ref: Optional[LocalObjectReference] = None
    target_portal: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            iscsi=_ModelISCSIVolumeSource(
                chap_auth_discovery=self.chap_auth_discovery,
                chap_auth_session=self.chap_auth_discovery,
                fs_type=self.fs_type,
                initiator_name=self.initiator_name,
                iqn=self.iqn,
                iscsi_interface=self.iscsi_interface,
                lun=self.lun,
                portals=self.portals,
                read_only=self.read_only,
                secret_ref=self.secret_ref,
                target_portal=self.target_portal,
            ),
        )

chap_auth_discovery

chap_auth_discovery: Optional[bool] = None

chap_auth_session

chap_auth_session: Optional[bool] = None

fs_type

fs_type: Optional[str] = None

initiator_name

initiator_name: Optional[str] = None

iqn

iqn: str

iscsi_interface

iscsi_interface: Optional[str] = None

lun

lun: int

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

portals

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

target_portal

target_portal: str

PhotonPersistentDiskVolume

A Photon Persistent Disk representation.

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

@dataclass(kw_only=True)
class PhotonPersistentDiskVolume(_BaseVolume):
    """A Photon Persistent Disk representation."""

    fs_type: Optional[str] = None
    pd_id: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            photon_persistent_disk=_ModelPhotonPersistentDiskVolumeSource(fs_type=self.fs_type, pd_id=self.pd_id),
        )

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

pd_id

pd_id: 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

PortworxVolume

PortworxVolume represents a Portworx volume to mount to a container.

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

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

    fs_type: Optional[str] = None
    volume_id: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            portworx_volume=_ModelPortworxVolumeSource(
                fs_type=self.fs_type, read_only=self.read_only, volume_id=self.volume_id
            ),
        )

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

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_id

volume_id: str

QuobyteVolume

QuobyteVolume represents a Quobyte volume to mount to a container.

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

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

    group: Optional[str] = None
    registry: str
    tenant: Optional[str] = None
    user: Optional[str] = None
    volume: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            quobyte=_ModelQuobyteVolumeSource(
                group=self.group,
                read_only=self.read_only,
                registry=self.registry,
                tenant=self.tenant,
                user=self.user,
                volume=self.volume,
            ),
        )

group

group: 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

registry

registry: str

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

tenant

tenant: Optional[str] = None

user

user: Optional[str] = None

volume

volume: str

RBDVolume

An RDB volume representation.

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

@dataclass(kw_only=True)
class RBDVolume(_BaseVolume):
    """An RDB volume representation."""

    fs_type: Optional[str] = None
    image: str
    keyring: Optional[str] = None
    monitors: List[str]
    pool: Optional[str] = None
    secret_ref: Optional[LocalObjectReference] = None
    user: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            rbd=_ModelRBDVolumeSource(
                fs_type=self.fs_type,
                image=self.image,
                keyring=self.keyring,
                monitors=self.monitors,
                pool=self.pool,
                read_only=self.read_only,
                secret_ref=self.secret_ref,
                user=self.user,
            ),
        )

fs_type

fs_type: Optional[str] = None

image

image: str

keyring

keyring: Optional[str] = None

monitors

monitors: List[str]

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

pool

pool: Optional[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

user

user: Optional[str] = None

ScaleIOVolume

ScaleIOVolume represents a ScaleIO volume to mount to the container.

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

@dataclass(kw_only=True)
class ScaleIOVolume(_BaseVolume):
    """`ScaleIOVolume` represents a ScaleIO volume to mount to the container."""

    fs_type: Optional[str] = None
    gateway: str
    protection_domain: Optional[str] = None
    secret_ref: LocalObjectReference
    ssl_enabled: Optional[bool] = None
    storage_mode: Optional[str] = None
    storage_pool: Optional[str] = None
    system: str
    volume_name: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            scale_io=_ModelScaleIOVolumeSource(
                fs_type=self.fs_type,
                gateway=self.gateway,
                protection_domain=self.protection_domain,
                read_only=self.read_only,
                secret_ref=self.secret_ref,
                ssl_enabled=self.ssl_enabled,
                storage_mode=self.storage_mode,
                storage_pool=self.storage_pool,
                system=self.system,
                volume_name=self.volume_name,
            ),
        )

fs_type

fs_type: Optional[str] = None

gateway

gateway: str

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

protection_domain

protection_domain: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

secret_ref

secret_ref: LocalObjectReference

ssl_enabled

ssl_enabled: Optional[bool] = None

storage_mode

storage_mode: Optional[str] = None

storage_pool

storage_pool: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

system

system: str

volume_name

volume_name: Optional[str] = None

StorageOSVolume

StorageOSVolume represents a Storage OS volume to mount.

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

@dataclass(kw_only=True)
class StorageOSVolume(_BaseVolume):
    """`StorageOSVolume` represents a Storage OS volume to mount."""

    fs_type: Optional[str] = None
    secret_ref: Optional[LocalObjectReference] = None
    volume_name: Optional[str] = None
    volume_namespace: Optional[str] = None

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            storageos=_ModelStorageOSVolumeSource(
                fs_type=self.fs_type,
                read_only=self.read_only,
                secret_ref=self.secret_ref,
                volume_name=self.volume_name,
                volume_namespace=self.volume_namespace,
            ),
        )

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

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

volume_name

volume_name: Optional[str] = None

volume_namespace

volume_namespace: Optional[str] = None

VsphereVirtualDiskVolume

VsphereVirtualDiskVolume represents a vSphere virtual disk volume to mount.

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

@dataclass(kw_only=True)
class VsphereVirtualDiskVolume(_BaseVolume):
    """`VsphereVirtualDiskVolume` represents a vSphere virtual disk volume to mount."""

    fs_type: Optional[str] = None
    storage_policy_id: Optional[str] = None
    storage_policy_name: Optional[str] = None
    volume_path: str

    def _build_volume(self) -> _ModelVolume:
        assert self.name
        return _ModelVolume(
            name=self.name,
            vsphere_volume=_ModelVsphereVirtualDiskVolumeSource(
                fs_type=self.fs_type,
                storage_policy_id=self.storage_policy_id,
                storage_policy_name=self.storage_policy_name,
                volume_path=self.volume_path,
            ),
        )

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

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

storage_policy_id

storage_policy_id: Optional[str] = None

storage_policy_name

storage_policy_name: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

volume_path

volume_path: str

Comments