Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.jobs.runners.util.cli.job

"""
Abstract base class for cli job plugins.
"""

from abc import (
    ABCMeta,
    abstractmethod,
)
from enum import Enum
from typing import (
    Dict,
    List,
)

from typing_extensions import TypeAlias

try:
    from galaxy.model import Job

    job_states: TypeAlias = Job.states
except ImportError:
    # Not in Galaxy, map Galaxy job states to Pulsar ones.
    class job_states(str, Enum):  # type: ignore[no-redef]
        RUNNING = "running"
        OK = "complete"
        QUEUED = "queued"
        ERROR = "failed"


[docs]class BaseJobExec(metaclass=ABCMeta):
[docs] def __init__(self, **params): """ Constructor for CLI job executor. """ self.params = params.copy()
[docs] def job_script_kwargs(self, ofile, efile, job_name): """Return extra keyword argument for consumption by job script module. """ return {}
[docs] @abstractmethod def submit(self, script_file): """ Given specified script_file path, yield command to submit it to external job manager. """
[docs] @abstractmethod def delete(self, job_id): """ Given job id, return command to stop execution or dequeue specified job. """
[docs] @abstractmethod def get_status(self, job_ids=None): """ Return command to get statuses of specified job ids. """
[docs] @abstractmethod def get_single_status(self, job_id): """ Return command to get the status of a single, specified job. """
[docs] @abstractmethod def parse_status(self, status: str, job_ids: List[str]) -> Dict[str, job_states]: """ Parse the statuses of output from get_status command. """
[docs] @abstractmethod def parse_single_status(self, status: str, job_id: str) -> job_states: """ Parse the status of output from get_single_status command. """
[docs] def get_failure_reason(self, job_id): """ Return the failure reason for the given job_id. """ return None
[docs] def parse_failure_reason(self, reason, job_id): """ Parses the failure reason, assigning it against a """ return None
__all__ = ( "BaseJobExec", "job_states", )