diff --git a/lib/galaxy/jobs/runners/util/cli/job/__init__.py b/lib/galaxy/jobs/runners/util/cli/job/__init__.py
index 000c1f38942b..2556df5d2aec 100644
--- a/lib/galaxy/jobs/runners/util/cli/job/__init__.py
+++ b/lib/galaxy/jobs/runners/util/cli/job/__init__.py
@@ -6,6 +6,10 @@
     abstractmethod,
 )
 from enum import Enum
+from typing import (
+    Dict,
+    List,
+)
 
 try:
     from galaxy.model import Job
@@ -60,7 +64,7 @@ def get_single_status(self, job_id):
         """
 
     @abstractmethod
-    def parse_status(self, status, job_ids):
+    def parse_status(self, status: str, job_ids: List[str]) -> Dict[str, str]:
         """
         Parse the statuses of output from get_status command.
         """
diff --git a/lib/galaxy/jobs/runners/util/cli/job/lsf.py b/lib/galaxy/jobs/runners/util/cli/job/lsf.py
index cdaedf447a42..2a5903e4b486 100644
--- a/lib/galaxy/jobs/runners/util/cli/job/lsf.py
+++ b/lib/galaxy/jobs/runners/util/cli/job/lsf.py
@@ -99,7 +99,7 @@ def parse_failure_reason(self, reason, job_id):
                 return runner_states.WALLTIME_REACHED
         return None
 
-    def _get_job_state(self, state):
+    def _get_job_state(self, state: str) -> str:
         # based on:
         # https://www.ibm.com/support/knowledgecenter/en/SSETD4_9.1.3/lsf_admin/job_state_lsf.html
         # https://www.ibm.com/support/knowledgecenter/en/SSETD4_9.1.2/lsf_command_ref/bjobs.1.html
@@ -115,7 +115,7 @@ def _get_job_state(self, state):
                 "UNKWN": job_states.ERROR,
                 "WAIT": job_states.QUEUED,
                 "ZOMBI": job_states.ERROR,
-            }.get(state)
+            }[state]
         except KeyError:
             raise KeyError(f"Failed to map LSF status code [{state}] to job state.")
 
diff --git a/lib/galaxy/jobs/runners/util/cli/job/slurm.py b/lib/galaxy/jobs/runners/util/cli/job/slurm.py
index 5c1626d7315f..66eda28fb28c 100644
--- a/lib/galaxy/jobs/runners/util/cli/job/slurm.py
+++ b/lib/galaxy/jobs/runners/util/cli/job/slurm.py
@@ -64,7 +64,7 @@ def parse_single_status(self, status, job_id):
         # else line like "slurm_load_jobs error: Invalid job id specified"
         return job_states.OK
 
-    def _get_job_state(self, state):
+    def _get_job_state(self, state: str) -> str:
         try:
             return {
                 "F": job_states.ERROR,
@@ -72,7 +72,7 @@ def _get_job_state(self, state):
                 "CG": job_states.RUNNING,
                 "PD": job_states.QUEUED,
                 "CD": job_states.OK,
-            }.get(state)
+            }[state]
         except KeyError:
             raise KeyError(f"Failed to map slurm status code [{state}] to job state.")
 
diff --git a/lib/galaxy/jobs/runners/util/cli/job/torque.py b/lib/galaxy/jobs/runners/util/cli/job/torque.py
index db9f1df15e29..1e7bad382789 100644
--- a/lib/galaxy/jobs/runners/util/cli/job/torque.py
+++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py
@@ -1,8 +1,4 @@
 from logging import getLogger
-from typing import (
-    Dict,
-    Union,
-)
 
 from galaxy.util import parse_xml_string
 from . import (
@@ -68,7 +64,7 @@ def get_status(self, job_ids=None):
     def get_single_status(self, job_id):
         return f"qstat -f {job_id}"
 
-    def parse_status(self, status: str, job_ids) -> Union[Dict, None]:
+    def parse_status(self, status, job_ids):
         # in case there's noise in the output, find the big blob 'o xml
         tree = None
         rval = {}
@@ -81,7 +77,7 @@ def parse_status(self, status: str, job_ids) -> Union[Dict, None]:
                 tree = None
         if tree is None:
             log.warning(f"No valid qstat XML return from `qstat -x`, got the following: {status}")
-            return None
+            return {}
         else:
             for job in tree.findall("Job"):
                 job_id_elem = job.find("Job_Id")
@@ -91,6 +87,7 @@ def parse_status(self, status: str, job_ids) -> Union[Dict, None]:
                     job_state_elem = job.find("job_state")
                     assert job_state_elem is not None
                     state = job_state_elem.text
+                    assert state
                     # map PBS job states to Galaxy job states.
                     rval[id_] = self._get_job_state(state)
         return rval
@@ -103,11 +100,9 @@ def parse_single_status(self, status, job_id):
         # no state found, job has exited
         return job_states.OK
 
-    def _get_job_state(self, state):
+    def _get_job_state(self, state: str) -> str:
         try:
-            return {"E": job_states.RUNNING, "R": job_states.RUNNING, "Q": job_states.QUEUED, "C": job_states.OK}.get(
-                state
-            )
+            return {"E": job_states.RUNNING, "R": job_states.RUNNING, "Q": job_states.QUEUED, "C": job_states.OK}[state]
         except KeyError:
             raise KeyError(f"Failed to map torque status code [{state}] to job state.")