mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 00:12:08 +08:00
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.
This commit is contained in:
@@ -11,10 +11,12 @@ from typing import (
|
||||
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 @@ class BaseJobExec(metaclass=ABCMeta):
|
||||
"""
|
||||
|
||||
@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.
|
||||
"""
|
||||
|
||||
@@ -100,7 +100,7 @@ class Torque(BaseJobExec):
|
||||
# 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:
|
||||
|
||||
@@ -111,6 +111,7 @@ from sqlalchemy.sql import exists
|
||||
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]
|
||||
|
||||
Reference in New Issue
Block a user