Workflow

Workflow

The base Workflow class for Hera.

Workflow implements the contextmanager interface so allows usage of with, under which any hera.workflows.protocol.Templatable object instantiated under the context will be added to the Workflow’s list of templates.

Workflows can be created directly on your Argo cluster via create. They can also be dumped to yaml via to_yaml or built according to the Argo schema via build to get an OpenAPI model object.

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

class Workflow(
    ArgumentsMixin,
    HookMixin,
    VolumeMixin,
    MetricsMixin,
    ModelMapperMixin,
    TemplateDecoratorFuncsMixin,
):
    """The base Workflow class for Hera.

    Workflow implements the contextmanager interface so allows usage of `with`, under which
    any `hera.workflows.protocol.Templatable` object instantiated under the context will be
    added to the Workflow's list of templates.

    Workflows can be created directly on your Argo cluster via `create`. They can also be dumped
    to yaml via `to_yaml` or built according to the Argo schema via `build` to get an OpenAPI model
    object.
    """

    def _build_volume_claim_templates(self) -> Optional[List]:
        return ((self.volume_claim_templates or []) + (self._build_persistent_volume_claims() or [])) or None

    def _build_on_exit(self) -> Optional[str]:
        if isinstance(self.on_exit, Templatable):
            return self.on_exit._build_template().name  # type: ignore
        return self.on_exit

    def _build_templates(self) -> Optional[List[TTemplate]]:
        """Builds the templates into an Argo schema."""
        templates = []
        for template in self.templates:
            if isinstance(template, HookMixin):
                template = template._dispatch_hooks()

            if isinstance(template, Templatable):
                templates.append(template._build_template())
            elif isinstance(template, get_args(TTemplate)):
                templates.append(template)
            else:
                raise InvalidType(f"{type(template)} is not a valid template type")

            if isinstance(template, VolumeClaimable):
                claims = template._build_persistent_volume_claims()  # type: ignore
                # If there are no claims, continue, nothing to add
                if not claims:
                    continue
                # If there are no volume claim templates, set them to the constructed claims
                elif self.volume_claim_templates is None:
                    self.volume_claim_templates = claims
                else:
                    # otherwise, we need to merge the two lists of volume claim templates. This prioritizes the
                    # already existing volume claim templates under the assumption that the user has already set
                    # a claim template on the workflow intentionally, or the user is sharing the same volumes across
                    # different templates
                    current_volume_claims_map = {}
                    for claim in self.volume_claim_templates:
                        assert claim.metadata is not None, "expected a workflow volume claim with metadata"
                        assert claim.metadata.name is not None, "expected a named workflow volume claim"
                        current_volume_claims_map[claim.metadata.name] = claim

                    new_volume_claims_map = {}
                    for claim in claims:
                        assert claim.metadata is not None, "expected a volume claim with metadata"
                        assert claim.metadata.name is not None, "expected a named volume claim"
                        new_volume_claims_map[claim.metadata.name] = claim

                    for claim_name, claim in new_volume_claims_map.items():
                        if claim_name not in current_volume_claims_map:
                            self.volume_claim_templates.append(claim)
        return templates or None

    # Workflow fields - https://argoproj.github.io/argo-workflows/fields/#workflow
    api_version: Annotated[Optional[str], _WorkflowModelMapper("api_version")] = None
    kind: Annotated[Optional[str], _WorkflowModelMapper("kind")] = None
    status: Annotated[Optional[_ModelWorkflowStatus], _WorkflowModelMapper("status")] = None

    # ObjectMeta fields - https://argoproj.github.io/argo-workflows/fields/#objectmeta
    annotations: Annotated[Optional[Dict[str, str]], _WorkflowModelMapper("metadata.annotations")] = None
    creation_timestamp: Annotated[Optional[Time], _WorkflowModelMapper("metadata.creation_timestamp")] = None
    deletion_grace_period_seconds: Annotated[
        Optional[int], _WorkflowModelMapper("metadata.deletion_grace_period_seconds")
    ] = None
    deletion_timestamp: Annotated[Optional[Time], _WorkflowModelMapper("metadata.deletion_timestamp")] = None
    finalizers: Annotated[Optional[List[str]], _WorkflowModelMapper("metadata.finalizers")] = None
    generate_name: Annotated[Optional[str], _WorkflowModelMapper("metadata.generate_name")] = None
    generation: Annotated[Optional[int], _WorkflowModelMapper("metadata.generation")] = None
    labels: Annotated[Optional[Dict[str, str]], _WorkflowModelMapper("metadata.labels")] = None
    managed_fields: Annotated[Optional[List[ManagedFieldsEntry]], _WorkflowModelMapper("metadata.managed_fields")] = (
        None
    )
    name: Annotated[Optional[str], _WorkflowModelMapper("metadata.name")] = None
    namespace: Annotated[Optional[str], _WorkflowModelMapper("metadata.namespace")] = None
    owner_references: Annotated[Optional[List[OwnerReference]], _WorkflowModelMapper("metadata.owner_references")] = (
        None
    )
    resource_version: Annotated[Optional[str], _WorkflowModelMapper("metadata.resource_version")] = None
    self_link: Annotated[Optional[str], _WorkflowModelMapper("metadata.self_link")] = None
    uid: Annotated[Optional[str], _WorkflowModelMapper("metadata.uid")] = None

    # WorkflowSpec fields - https://argoproj.github.io/argo-workflows/fields/#workflowspec
    active_deadline_seconds: Annotated[Optional[int], _WorkflowModelMapper("spec.active_deadline_seconds")] = None
    affinity: Annotated[Optional[Affinity], _WorkflowModelMapper("spec.affinity")] = None
    archive_logs: Annotated[Optional[bool], _WorkflowModelMapper("spec.archive_logs")] = None
    artifact_gc: Annotated[Optional[ArtifactGC], _WorkflowModelMapper("spec.artifact_gc")] = None
    artifact_repository_ref: Annotated[
        Optional[ArtifactRepositoryRef], _WorkflowModelMapper("spec.artifact_repository_ref")
    ] = None
    automount_service_account_token: Annotated[
        Optional[bool], _WorkflowModelMapper("spec.automount_service_account_token")
    ] = None
    dns_config: Annotated[Optional[PodDNSConfig], _WorkflowModelMapper("spec.dns_config")] = None
    dns_policy: Annotated[Optional[str], _WorkflowModelMapper("spec.dns_policy")] = None
    entrypoint: Annotated[Optional[str], _WorkflowModelMapper("spec.entrypoint")] = None
    executor: Annotated[Optional[ExecutorConfig], _WorkflowModelMapper("spec.executor")] = None
    hooks: Annotated[Optional[Dict[str, LifecycleHook]], _WorkflowModelMapper("spec.hooks")] = None
    host_aliases: Annotated[Optional[List[HostAlias]], _WorkflowModelMapper("spec.host_aliases")] = None
    host_network: Annotated[Optional[bool], _WorkflowModelMapper("spec.host_network")] = None
    image_pull_secrets: Annotated[ImagePullSecretsT, _WorkflowModelMapper("spec.image_pull_secrets")] = None
    node_selector: Annotated[Optional[Dict[str, str]], _WorkflowModelMapper("spec.node_selector")] = None
    on_exit: Annotated[Optional[Union[str, Templatable]], _WorkflowModelMapper("spec.on_exit", _build_on_exit)] = None
    parallelism: Annotated[Optional[int], _WorkflowModelMapper("spec.parallelism")] = None
    pod_disruption_budget: Annotated[
        Optional[PodDisruptionBudgetSpec], _WorkflowModelMapper("spec.pod_disruption_budget")
    ] = None
    pod_gc: Annotated[Optional[PodGC], _WorkflowModelMapper("spec.pod_gc")] = None
    pod_metadata: Annotated[Optional[Metadata], _WorkflowModelMapper("spec.pod_metadata")] = None
    pod_priority: Annotated[Optional[int], _WorkflowModelMapper("spec.pod_priority")] = None
    pod_priority_class_name: Annotated[Optional[str], _WorkflowModelMapper("spec.pod_priority_class_name")] = None
    pod_spec_patch: Annotated[Optional[str], _WorkflowModelMapper("spec.pod_spec_patch")] = None
    priority: Annotated[Optional[int], _WorkflowModelMapper("spec.priority")] = None
    retry_strategy: Annotated[Optional[RetryStrategy], _WorkflowModelMapper("spec.retry_strategy")] = None
    scheduler_name: Annotated[Optional[str], _WorkflowModelMapper("spec.scheduler_name")] = None
    security_context: Annotated[Optional[PodSecurityContext], _WorkflowModelMapper("spec.security_context")] = None
    service_account_name: Annotated[Optional[str], _WorkflowModelMapper("spec.service_account_name")] = None
    shutdown: Annotated[Optional[str], _WorkflowModelMapper("spec.shutdown")] = None
    suspend: Annotated[Optional[bool], _WorkflowModelMapper("spec.suspend")] = None
    synchronization: Annotated[Optional[Synchronization], _WorkflowModelMapper("spec.synchronization")] = None
    template_defaults: Annotated[Optional[_ModelTemplate], _WorkflowModelMapper("spec.template_defaults")] = None
    templates: Annotated[
        List[Union[_ModelTemplate, Templatable]], _WorkflowModelMapper("spec.templates", _build_templates)
    ] = []
    tolerations: Annotated[Optional[List[Toleration]], _WorkflowModelMapper("spec.tolerations")] = None
    ttl_strategy: Annotated[Optional[TTLStrategy], _WorkflowModelMapper("spec.ttl_strategy")] = None
    volume_claim_gc: Annotated[Optional[VolumeClaimGC], _WorkflowModelMapper("spec.volume_claim_gc")] = None
    volume_claim_templates: Annotated[
        Optional[List[PersistentVolumeClaim]],
        _WorkflowModelMapper("spec.volume_claim_templates", _build_volume_claim_templates),
    ] = None
    workflow_metadata: Annotated[Optional[WorkflowMetadata], _WorkflowModelMapper("spec.workflow_metadata")] = None
    workflow_template_ref: Annotated[
        Optional[WorkflowTemplateRef], _WorkflowModelMapper("spec.workflow_template_ref")
    ] = None

    # Override types for mixin fields
    arguments: Annotated[
        ArgumentsT,
        _WorkflowModelMapper("spec.arguments", ArgumentsMixin._build_arguments),
    ] = None
    metrics: Annotated[
        MetricsT,
        _WorkflowModelMapper("spec.metrics", MetricsMixin._build_metrics),
    ] = None
    volumes: Annotated[VolumesT, _WorkflowModelMapper("spec.volumes", VolumeMixin._build_volumes)] = None

    # Hera-specific fields
    workflows_service: Optional[WorkflowsService] = None

    @validator("name", pre=True, always=True)
    def _set_name(cls, v):
        if v is not None and len(v) > NAME_LIMIT:
            raise ValueError(f"name must be no more than {NAME_LIMIT} characters: {v}")
        return v

    @validator("generate_name", pre=True, always=True)
    def _set_generate_name(cls, v):
        if v is not None and len(v) > NAME_LIMIT:
            raise ValueError(f"generate_name must be no more than {NAME_LIMIT} characters: {v}")
        return v

    @validator("api_version", pre=True, always=True)
    def _set_api_version(cls, v):
        if v is None:
            return global_config.api_version
        return v

    @validator("workflows_service", pre=True, always=True)
    def _set_workflows_service(cls, v):
        if v is None:
            return WorkflowsService()
        return v

    @validator("kind", pre=True, always=True)
    def _set_kind(cls, v):
        if v is None:
            return cls.__name__  # type: ignore
        return v

    @validator("namespace", pre=True, always=True)
    def _set_namespace(cls, v):
        if v is None:
            return global_config.namespace
        return v

    @validator("service_account_name", pre=True, always=True)
    def _set_service_account_name(cls, v):
        if v is None:
            return global_config.service_account_name
        return v

    @validator("image_pull_secrets", pre=True, always=True)
    def _set_image_pull_secrets(cls, v):
        if v is None:
            return None

        if isinstance(v, str):
            return [LocalObjectReference(name=v)]
        elif isinstance(v, LocalObjectReference):
            return [v]

        assert isinstance(v, list), (
            "`image_pull_secrets` expected to be either a `str`, a `LocalObjectReferences`, a list of `str`, "
            "or a list of `LocalObjectReferences`"
        )
        result = []
        for secret in v:
            if isinstance(secret, str):
                result.append(LocalObjectReference(name=secret))
            elif isinstance(secret, LocalObjectReference):
                result.append(secret)
        return result

    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}}}}}")

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

        model_workflow = _ModelWorkflow(
            metadata=ObjectMeta(),
            spec=_ModelWorkflowSpec(),
        )
        return _WorkflowModelMapper.build_model(Workflow, self, model_workflow)

    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)

    def __eq__(self, other) -> bool:
        """Verifies equality of `self` with the specified `other`."""
        if other.__class__ is self.__class__:
            return self.to_dict() == other.to_dict()

        return False

    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)

    def create(self, wait: bool = False, poll_interval: int = 5) -> TWorkflow:
        """Creates the Workflow on the Argo cluster.

        Parameters
        ----------
        wait: bool = False
            If false then the workflow is created asynchronously and the function returns immediately.
            If true then the workflow is created and the function blocks until the workflow is done executing.
        poll_interval: int = 5
            The interval in seconds to poll the workflow status if wait is true. Ignored when wait is false.
        """
        assert self.workflows_service, "workflow service not initialized"
        assert self.namespace, "workflow namespace not defined"

        wf = self.workflows_service.create_workflow(
            WorkflowCreateRequest(workflow=self.build()),  # type: ignore
            namespace=self.namespace,
        )
        # set the workflow name to the name returned by the API, which helps cover the case of users relying on
        # `generate_name=True`
        self.name = wf.metadata.name

        if wait:
            return self.wait(poll_interval=poll_interval)
        return wf

    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 self.workflows_service is not None, "workflow 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.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

    def lint(self) -> TWorkflow:
        """Lints the Workflow using the Argo cluster."""
        assert self.workflows_service, "workflow service not initialized"
        assert self.namespace, "workflow namespace not defined"
        return self.workflows_service.lint_workflow(
            WorkflowLintRequest(workflow=self.build()),  # type: ignore
            namespace=self.namespace,
        )

    def _add_sub(self, node: Any):
        """Adds the given node (expected to satisfy the `Templatable` protocol) to the context."""
        if not isinstance(node, (Templatable, _ModelTemplate)):
            raise InvalidType(type(node))
        self.templates.append(node)

    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()

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

        Examples:
            >>> my_workflow = Workflow(name="my-workflow")
            >>> my_workflow == Workflow.from_dict(my_workflow.to_dict())
            True
        """
        return cls._from_dict(model_dict, _ModelWorkflow)

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

        Examples:
            >>> my_workflow = Workflow.from_yaml(yaml_str)
        """
        return cls._from_yaml(yaml_str, _ModelWorkflow)

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

        Examples:
            >>> yaml_file = Path(...)
            >>> my_workflow = Workflow.from_file(yaml_file)
        """
        return cls._from_file(yaml_file, _ModelWorkflow)

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

    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

    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)

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

artifact_repository_ref

artifact_repository_ref: Optional[ArtifactRepositoryRef] = None

automount_service_account_token

automount_service_account_token: Optional[bool] = None

creation_timestamp

creation_timestamp: Optional[Time] = 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

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

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

scheduler_name

scheduler_name: Optional[str] = None

security_context

security_context: Optional[PodSecurityContext] = None

self_link

self_link: Optional[str] = None

service_account_name

service_account_name: Optional[str] = None

shutdown

shutdown: Optional[str] = None

status

status: Optional[WorkflowStatus] = None

suspend

suspend: Optional[bool] = None

synchronization

synchronization: Optional[Synchronization] = None

template_defaults

template_defaults: Optional[Template] = None

templates

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

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

workflow_metadata

workflow_metadata: Optional[WorkflowMetadata] = None

workflow_template_ref

workflow_template_ref: Optional[WorkflowTemplateRef] = None

workflows_service

workflows_service: Optional[WorkflowsService] = 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="docker/whalesay", command=["cowsay", "foo"])

See how to use it in the Container example.

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

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="docker/whalesay", 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=self.priority,
            priority_class_name=self.priority_class_name,
            retry_strategy=self.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

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[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

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

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

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

priority_class_name

priority_class_name: Optional[str] = None

retry_strategy

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

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

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

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:
        assert isinstance(hera_obj, ModelMapperMixin)

        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

@classmethod
def build_model(
    cls, hera_class: Type[ModelMapperMixin], hera_obj: ModelMapperMixin, model: TWorkflow
) -> TWorkflow:
    assert isinstance(hera_obj, ModelMapperMixin)

    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

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

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

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[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

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

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

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

priority_class_name

priority_class_name: Optional[str] = None

retry_strategy

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

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

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

def parallel(self) -> Parallel:
    """Returns a Parallel object which can be used in a sub-context manager."""
    return Parallel()

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

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)

build

build() -> TWorkflow

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

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

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

    model_workflow = _ModelWorkflow(
        metadata=ObjectMeta(),
        spec=_ModelWorkflowSpec(),
    )
    return _WorkflowModelMapper.build_model(Workflow, self, model_workflow)

container

container(**container_kwargs) -> Callable

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

@_add_type_hints(Container)  # type: ignore
def container(self, **container_kwargs) -> Callable:
    self._check_if_enabled("container")
    from hera.workflows.container import Container

    def container_decorator(func: Callable[FuncIns, FuncR]) -> Callable:
        name = container_kwargs.pop("name", func.__name__.replace("_", "-"))

        signature = inspect.signature(func)
        func_inputs = signature.parameters
        inputs = []
        if len(func_inputs) >= 1:
            input_arg = list(func_inputs.values())[0].annotation
            if issubclass(input_arg, (InputV1, InputV2)):
                inputs = input_arg._get_inputs(add_missing_path=True)

        func_return = signature.return_annotation
        outputs = []
        if func_return and issubclass(func_return, (OutputV1, OutputV2)):
            outputs = func_return._get_outputs(add_missing_path=True)

        # Open context to add `Container` object automatically
        with self:
            container_template = Container(
                name=name,
                inputs=inputs,
                outputs=outputs,
                **container_kwargs,
            )

        @functools.wraps(func)
        def container_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 = container_template.name  # type: ignore

                    assert isinstance(subnode_name, str)
                    subnode_name = subnode_name.replace("_", "-")

                return self._create_subnode(subnode_name, func, container_template, *args, **kwargs)

            if _context.pieces:
                return container_template.__call__(*args, **kwargs)

            # Do not allow kwargs
            return func(*args)  # type: ignore

        # Set the template name to the inferred name
        container_call_wrapper.template_name = name  # type: ignore

        return container_call_wrapper

    return container_decorator

create

create(wait: bool = False, poll_interval: int = 5) -> TWorkflow

Creates the Workflow on the Argo cluster.

Parameters

wait: bool = False If false then the workflow is created asynchronously and the function returns immediately. If true then the workflow is created and the function blocks until the workflow is done executing. poll_interval: int = 5 The interval in seconds to poll the workflow status if wait is true. Ignored when wait is false.

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

def create(self, wait: bool = False, poll_interval: int = 5) -> TWorkflow:
    """Creates the Workflow on the Argo cluster.

    Parameters
    ----------
    wait: bool = False
        If false then the workflow is created asynchronously and the function returns immediately.
        If true then the workflow is created and the function blocks until the workflow is done executing.
    poll_interval: int = 5
        The interval in seconds to poll the workflow status if wait is true. Ignored when wait is false.
    """
    assert self.workflows_service, "workflow service not initialized"
    assert self.namespace, "workflow namespace not defined"

    wf = self.workflows_service.create_workflow(
        WorkflowCreateRequest(workflow=self.build()),  # type: ignore
        namespace=self.namespace,
    )
    # set the workflow name to the name returned by the API, which helps cover the case of users relying on
    # `generate_name=True`
    self.name = wf.metadata.name

    if wait:
        return self.wait(poll_interval=poll_interval)
    return wf

dag

dag(**dag_kwargs) -> Callable

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

@_add_type_hints(DAG)  # type: ignore
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)

from_dict

from_dict(model_dict: Dict) -> ModelMapperMixin

Create a Workflow from a Workflow contained in a dict.

Examples:

>>> my_workflow = Workflow(name="my-workflow")
>>> my_workflow == Workflow.from_dict(my_workflow.to_dict())
True

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

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

    Examples:
        >>> my_workflow = Workflow(name="my-workflow")
        >>> my_workflow == Workflow.from_dict(my_workflow.to_dict())
        True
    """
    return cls._from_dict(model_dict, _ModelWorkflow)

from_file

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

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

Examples:

>>> yaml_file = Path(...)
>>> my_workflow = Workflow.from_file(yaml_file)

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

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

    Examples:
        >>> yaml_file = Path(...)
        >>> my_workflow = Workflow.from_file(yaml_file)
    """
    return cls._from_file(yaml_file, _ModelWorkflow)

from_yaml

from_yaml(yaml_str: str) -> ModelMapperMixin

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

Examples:

>>> my_workflow = Workflow.from_yaml(yaml_str)

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

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

    Examples:
        >>> my_workflow = Workflow.from_yaml(yaml_str)
    """
    return cls._from_yaml(yaml_str, _ModelWorkflow)

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

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

get_workflow_link() -> str

Returns the workflow link for the workflow.

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

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

lint

lint() -> TWorkflow

Lints the Workflow using the Argo cluster.

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

def lint(self) -> TWorkflow:
    """Lints the Workflow using the Argo cluster."""
    assert self.workflows_service, "workflow service not initialized"
    assert self.namespace, "workflow namespace not defined"
    return self.workflows_service.lint_workflow(
        WorkflowLintRequest(workflow=self.build()),  # type: ignore
        namespace=self.namespace,
    )

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

@_add_type_hints(Script)  # type: ignore
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

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

@_add_type_hints(Steps)  # type: ignore
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)

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

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

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

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)

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

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 self.workflows_service is not None, "workflow 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.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