diff --git a/lib/galaxy/jobs/runners/util/cli/job/__init__.py b/lib/galaxy/jobs/runners/util/cli/job/__init__.py index 2556df5d2aec..278d5c9834b8 100644 --- a/lib/galaxy/jobs/runners/util/cli/job/__init__.py +++ b/lib/galaxy/jobs/runners/util/cli/job/__init__.py @@ -64,13 +64,13 @@ def get_single_status(self, job_id): """ @abstractmethod - def parse_status(self, status: str, job_ids: List[str]) -> Dict[str, str]: + def parse_status(self, status: str, job_ids: List[str]) -> Dict[str, job_states]: """ Parse the statuses of output from get_status command. """ @abstractmethod - def parse_single_status(self, status, job_id): + def parse_single_status(self, status: str, job_id: str) -> job_states: """ Parse the status of output from get_single_status command. """ diff --git a/lib/galaxy/jobs/runners/util/cli/job/torque.py b/lib/galaxy/jobs/runners/util/cli/job/torque.py index 1e7bad382789..ef6168d7ad4c 100644 --- a/lib/galaxy/jobs/runners/util/cli/job/torque.py +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -100,7 +100,7 @@ def parse_single_status(self, status, job_id): # no state found, job has exited return job_states.OK - def _get_job_state(self, state: str) -> str: + def _get_job_state(self, state: str) -> job_states: try: return {"E": job_states.RUNNING, "R": job_states.RUNNING, "Q": job_states.QUEUED, "C": job_states.OK}[state] except KeyError: diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 34af7c1d5e81..a9e27f2cae8c 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -111,6 +111,7 @@ from typing_extensions import ( Literal, Protocol, + TypeAlias, TypedDict, ) @@ -1380,7 +1381,7 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable): _numeric_metric = JobMetricNumeric _text_metric = JobMetricText - states = JobState + states: TypeAlias = JobState # states that are not expected to change, except through admin action or re-scheduling terminal_states = [states.OK, states.ERROR, states.DELETED]