From dad746eeed65379e7c31c47b9b9bcdb67f129b34 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Mon, 29 Jan 2024 16:14:50 +0000 Subject: [PATCH] More specific type annotation for ``BaseJobExec.parse_status()`` Add type annotation to ``BaseJobExec.parse_single_status()`` Follow-up on https://github.com/galaxyproject/galaxy/pull/17367#discussion_r1469748812 . Needed to add ``TypeAlias`` to ``model.Job.states`` to fix: ``` lib/galaxy/jobs/runners/util/cli/job/__init__.py:67: error: Variable "galaxy.jobs.runners.util.cli.job.job_states" is not valid as a type [valid-type] def parse_status(self, status: str, job_ids: List[str]) -> Dict[str, job_states]: ^ ``` See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases : > You should always use ``TypeAlias`` to define a type alias in a class body > or inside a function. --- lib/galaxy/jobs/runners/util/cli/job/__init__.py | 8 +++++--- lib/galaxy/jobs/runners/util/cli/job/torque.py | 2 +- lib/galaxy/model/__init__.py | 3 ++- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/lib/galaxy/jobs/runners/util/cli/job/__init__.py b/lib/galaxy/jobs/runners/util/cli/job/__init__.py index 2556df5d2aec..477f0999b435 100644 --- a/lib/galaxy/jobs/runners/util/cli/job/__init__.py +++ b/lib/galaxy/jobs/runners/util/cli/job/__init__.py @@ -11,10 +11,12 @@ List, ) +from typing_extensions import TypeAlias + try: from galaxy.model import Job - job_states = Job.states + 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] @@ -64,13 +66,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]