Fix type annotation of code using XML etree

by adding lxml-stubs to our typecheck requirements.

lib/galaxy/datatypes/dataproviders/hierarchy.py

@@ -3,7 +3,10 @@
 """
 import logging
 
-from galaxy.util import etree
+from galaxy.util import (
+    Element,
+    etree,
+)
 from . import line
 
 _TODO = """
@@ -65,7 +68,6 @@ def matches_selector(self, element, selector=None):
         # TODO: add more flexibility here w/o re-implementing xpath
         # TODO: fails with '#' - browser thinks it's an anchor - use urlencode
         # TODO: need removal/replacement of etree namespacing here - then move to string match
-        Element = getattr(etree, "_Element", etree.Element)
         return bool((selector is None) or (isinstance(element, Element) and selector in element.tag))
 
     def element_as_dict(self, element):

lib/galaxy/dependencies/__init__.py

@@ -67,18 +67,15 @@ def load_job_config_dict(job_conf_dict):
                 job_conf_path = join(dirname(self.config_file), job_conf_path)
             if ".xml" in job_conf_path:
                 try:
-                    try:
-                        for plugin in parse_xml(job_conf_path).find("plugins").findall("plugin"):
+                    job_conf_tree = parse_xml(job_conf_path)
+                    plugins_elem = job_conf_tree.find("plugins")
+                    if plugins_elem:
+                        for plugin in plugins_elem.findall("plugin"):
                             if "load" in plugin.attrib:
                                 self.job_runners.append(plugin.attrib["load"])
-                    except OSError:
-                        pass
-                    try:
-                        for plugin in parse_xml(job_conf_path).findall('.//destination/param[@id="rules_module"]'):
-                            self.job_rule_modules.append(plugin.text)
-                    except OSError:
-                        pass
-                except etree.ParseError:
+                    for plugin in job_conf_tree.findall('.//destination/param[@id="rules_module"]'):
+                        self.job_rule_modules.append(plugin.text)
+                except (OSError, etree.ParseError):
                     pass
             else:
                 try:

lib/galaxy/dependencies/pinned-typecheck-requirements.txt

@@ -1,6 +1,7 @@
 annotated-types==0.6.0 ; python_version >= "3.8" and python_version < "3.13"
 cffi==1.16.0 ; python_version >= "3.8" and python_version < "3.13"
 cryptography==41.0.7 ; python_version >= "3.8" and python_version < "3.13"
+lxml-stubs==0.5.1 ; python_version >= "3.8" and python_version < "3.13"
 mypy-extensions==1.0.0 ; python_version >= "3.8" and python_version < "3.13"
 mypy==1.8.0 ; python_version >= "3.8" and python_version < "3.13"
 pycparser==2.21 ; python_version >= "3.8" and python_version < "3.13"

lib/galaxy/jobs/__init__.py

@@ -79,7 +79,6 @@
     ToolEvaluator,
 )
 from galaxy.util import (
-    etree,
     parse_xml_string,
     RWXRWXRWX,
     safe_makedirs,
@@ -643,7 +642,7 @@ def get_tool_resource_xml(self, tool_id, tool_type):
                             field_name, self.resource_parameters
                         )
                         raise KeyError(message)
-                    fields.append(etree.fromstring(self.resource_parameters[field_name]))
+                    fields.append(parse_xml_string(self.resource_parameters[field_name]))
 
                 if fields:
                     conditional_element = parse_xml_string(self.JOB_RESOURCE_CONDITIONAL_XML)

lib/galaxy/jobs/dynamic_tool_destination.py

@@ -1647,14 +1647,14 @@ def get_destination_list_from_job_config(job_config_location) -> set:
 
         # Add all destination IDs from the job configuration xml file
         for destination in job_conf.getroot().iter("destination"):
-            if isinstance(destination.get("id"), str):
-                destination_list.add(destination.get("id"))
-
+            destination_id = destination.get("id")
+            if destination_id:
+                destination_list.add(destination_id)
             else:
                 error = f"Destination ID '{str(destination)}"
                 error += "' in job configuration file cannot be"
                 error += " parsed. Things may not work as expected!"
-                log.debug(error)
+                log.warning(error)
 
     return destination_list
 

lib/galaxy/jobs/runners/util/cli/job/torque.py

@@ -1,4 +1,8 @@
 from logging import getLogger
+from typing import (
+    Dict,
+    Union,
+)
 
 from galaxy.util import parse_xml_string
 from . import (
@@ -64,7 +68,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, job_ids):
+    def parse_status(self, status: str, job_ids) -> Union[Dict, None]:
         # in case there's noise in the output, find the big blob 'o xml
         tree = None
         rval = {}
@@ -80,11 +84,15 @@ def parse_status(self, status, job_ids):
             return None
         else:
             for job in tree.findall("Job"):
-                id = job.find("Job_Id").text
-                if id in job_ids:
-                    state = job.find("job_state").text
+                job_id_elem = job.find("Job_Id")
+                assert job_id_elem
+                id_ = job_id_elem.text
+                if id_ in job_ids:
+                    job_state_elem = job.find("job_state")
+                    assert job_state_elem
+                    state = job_state_elem.text
                     # map PBS job states to Galaxy job states.
-                    rval[id] = self._get_job_state(state)
+                    rval[id_] = self._get_job_state(state)
         return rval
 
     def parse_single_status(self, status, job_id):

lib/galaxy/tool_shed/galaxy_install/tools/data_manager.py

@@ -111,8 +111,8 @@ def install_data_managers(
                     data_manager_id = elem.get("id", None)
                     if data_manager_id is None:
                         log.error(
-                            "A data manager was defined that does not have an id and will not be installed:\n%s"
-                            % xml_to_string(elem)
+                            "A data manager was defined that does not have an id and will not be installed:\n%s",
+                            xml_to_string(elem),
                         )
                         continue
                     data_manager_dict = (
@@ -170,7 +170,7 @@ def install_data_managers(
                     )
                     if data_manager:
                         rval.append(data_manager)
-                elif elem.tag is etree.Comment:
+                elif elem.tag is etree.Comment:  # type: ignore[comparison-overlap]
                     pass
                 else:
                     log.warning(f"Encountered unexpected element '{elem.tag}':\n{xml_to_string(elem)}")

lib/galaxy/tool_shed/galaxy_install/tools/tool_panel_manager.py

@@ -13,8 +13,9 @@
 from galaxy.tool_shed.util.repository_util import get_repository_owner
 from galaxy.tool_shed.util.shed_util_common import get_tool_panel_config_tool_path_install_dir
 from galaxy.util import (
-    etree,
+    Element,
     parse_xml_string,
+    SubElement,
     xml_to_string,
 )
 from galaxy.util.renamed_temporary_file import RenamedTemporaryFile
@@ -154,27 +155,27 @@ def config_elems_to_xml_file(self, config_elems, config_filename, tool_path, too
 
     def generate_tool_elem(
         self, tool_shed, repository_name, changeset_revision, owner, tool_file_path, tool, tool_section
-    ):
+    ) -> Element:
         """Create and return an ElementTree tool Element."""
         if tool_section is not None:
-            tool_elem = etree.SubElement(tool_section, "tool")
+            tool_elem = SubElement(tool_section, "tool")
         else:
-            tool_elem = etree.Element("tool")
+            tool_elem = Element("tool")
         tool_elem.attrib["file"] = tool_file_path
         if not tool.guid:
             raise ValueError("tool has no guid")
         tool_elem.attrib["guid"] = tool.guid
-        tool_shed_elem = etree.SubElement(tool_elem, "tool_shed")
+        tool_shed_elem = SubElement(tool_elem, "tool_shed")
         tool_shed_elem.text = tool_shed
-        repository_name_elem = etree.SubElement(tool_elem, "repository_name")
+        repository_name_elem = SubElement(tool_elem, "repository_name")
         repository_name_elem.text = repository_name
-        repository_owner_elem = etree.SubElement(tool_elem, "repository_owner")
+        repository_owner_elem = SubElement(tool_elem, "repository_owner")
         repository_owner_elem.text = owner
-        changeset_revision_elem = etree.SubElement(tool_elem, "installed_changeset_revision")
+        changeset_revision_elem = SubElement(tool_elem, "installed_changeset_revision")
         changeset_revision_elem.text = changeset_revision
-        id_elem = etree.SubElement(tool_elem, "id")
+        id_elem = SubElement(tool_elem, "id")
         id_elem.text = tool.id
-        version_elem = etree.SubElement(tool_elem, "version")
+        version_elem = SubElement(tool_elem, "version")
         version_elem.text = tool.version
         return tool_elem
 
@@ -297,7 +298,7 @@ def generate_tool_panel_elem_list(
         owner="",
     ):
         """Generate a list of ElementTree Element objects for each section or tool."""
-        elem_list: List[etree.Element] = []
+        elem_list: List[Element] = []
         tool_elem = None
         cleaned_repository_clone_url = remove_protocol_and_user_from_clone_url(repository_clone_url)
         if not owner:
@@ -335,6 +336,7 @@ def generate_tool_panel_elem_list(
                     tool_section if inside_section else None,
                 )
                 if inside_section:
+                    assert tool_section
                     if section_in_elem_list is not None:
                         elem_list[section_in_elem_list] = tool_section
                     else:
@@ -366,17 +368,13 @@ def generate_tool_section_dicts(self, tool_config=None, tool_sections=None) -> L
             tool_section_dicts.append(dict(tool_config=tool_config, id="", version="", name=""))
         return tool_section_dicts
 
-    def generate_tool_section_element_from_dict(self, tool_section_dict):
+    def generate_tool_section_element_from_dict(self, tool_section_dict: Dict[str, str]) -> Element:
         # The value of tool_section_dict looks like the following.
         # { id: <ToolSection id>, version : <ToolSection version>, name : <TooSection name>}
-        if tool_section_dict["id"]:
-            # Create a new tool section.
-            tool_section = etree.Element("section")
-            tool_section.attrib["id"] = tool_section_dict["id"]
-            tool_section.attrib["name"] = tool_section_dict["name"]
-            tool_section.attrib["version"] = tool_section_dict["version"]
-        else:
-            tool_section = None
+        tool_section = Element("section")
+        tool_section.attrib["id"] = tool_section_dict["id"]
+        tool_section.attrib["name"] = tool_section_dict["name"]
+        tool_section.attrib["version"] = tool_section_dict["version"]
         return tool_section
 
     def get_or_create_tool_section(self, toolbox, tool_panel_section_id, new_tool_panel_section_label=None):

lib/galaxy/tool_shed/metadata/metadata_generator.py

@@ -492,9 +492,9 @@ def generate_repository_dependency_metadata(self, repository_dependencies_config
             root = tree.getroot()
             xml_is_valid = root.tag == "repositories"
         if xml_is_valid:
-            invalid_repository_dependencies_dict = dict(description=root.get("description"))
+            invalid_repository_dependencies_dict: Dict[str, Any] = dict(description=root.get("description"))
             invalid_repository_dependency_tups = []
-            valid_repository_dependencies_dict = dict(description=root.get("description"))
+            valid_repository_dependencies_dict: Dict[str, Any] = dict(description=root.get("description"))
             valid_repository_dependency_tups = []
             for repository_elem in root.findall("repository"):
                 repository_dependency_tup, repository_dependency_is_valid, err_msg = self.handle_repository_elem(

lib/galaxy/tool_shed/tools/data_table_manager.py

@@ -9,7 +9,10 @@
 
 from galaxy.tool_shed.galaxy_install.client import InstallationTarget
 from galaxy.tool_shed.util import hg_util
-from galaxy.util import etree
+from galaxy.util import (
+    Element,
+    SubElement,
+)
 from galaxy.util.tool_shed import xml_util
 
 if TYPE_CHECKING:
@@ -29,24 +32,24 @@ def __init__(self, app: RequiredAppT):
 
     def generate_repository_info_elem(
         self, tool_shed: str, repository_name: str, changeset_revision: str, owner: str, parent_elem=None, **kwd
-    ) -> etree.Element:
+    ) -> Element:
         """Create and return an ElementTree repository info Element."""
         if parent_elem is None:
-            elem = etree.Element("tool_shed_repository")
+            elem = Element("tool_shed_repository")
         else:
-            elem = etree.SubElement(parent_elem, "tool_shed_repository")
-        tool_shed_elem = etree.SubElement(elem, "tool_shed")
+            elem = SubElement(parent_elem, "tool_shed_repository")
+        tool_shed_elem = SubElement(elem, "tool_shed")
         tool_shed_elem.text = tool_shed
-        repository_name_elem = etree.SubElement(elem, "repository_name")
+        repository_name_elem = SubElement(elem, "repository_name")
         repository_name_elem.text = repository_name
-        repository_owner_elem = etree.SubElement(elem, "repository_owner")
+        repository_owner_elem = SubElement(elem, "repository_owner")
         repository_owner_elem.text = owner
-        changeset_revision_elem = etree.SubElement(elem, "installed_changeset_revision")
+        changeset_revision_elem = SubElement(elem, "installed_changeset_revision")
         changeset_revision_elem.text = changeset_revision
         # add additional values
         # TODO: enhance additional values to allow e.g. use of dict values that will recurse
         for key, value in kwd.items():
-            new_elem = etree.SubElement(elem, key)
+            new_elem = SubElement(elem, key)
             new_elem.text = value
         return elem
 

lib/galaxy/tool_util/data/__init__.py

@@ -169,8 +169,9 @@ def __init__(
         filename: Optional[StrPath] = None,
         other_config_dict: Optional[StoresConfigFilePaths] = None,
     ) -> None:
-        self.name = config_element.get("name")
-        self.comment_char = config_element.get("comment_char")
+        name = config_element.get("name")
+        assert name
+        self.name = name
         self.empty_field_value = config_element.get("empty_field_value", "")
         self.empty_field_values: Dict[str, str] = {}
         self.allow_duplicate_entries = util.asbool(config_element.get("allow_duplicate_entries", True))
@@ -179,7 +180,7 @@ def __init__(
         self.tool_data_path = tool_data_path
         self.tool_data_path_files = tool_data_path_files
         self.other_config_dict = other_config_dict or {}
-        self.missing_index_file = None
+        self.missing_index_file: Optional[str] = None
         # increment this variable any time a new entry is added, or when the table is totally reloaded
         # This value has no external meaning, and does not represent an abstract version of the underlying data
         self._loaded_content_version = 1
@@ -348,11 +349,19 @@ def configure_and_load(
         # store repo info if available:
         repo_elem = config_element.find("tool_shed_repository")
         if repo_elem is not None:
+            tool_shed_elem = repo_elem.find("tool_shed")
+            assert tool_shed_elem
+            repository_name_elem = repo_elem.find("repository_name")
+            assert repository_name_elem
+            repository_owner_elem = repo_elem.find("repository_owner")
+            assert repository_owner_elem
+            installed_changeset_revision_elem = repo_elem.find("installed_changeset_revision")
+            assert installed_changeset_revision_elem
             repo_info = dict(
-                tool_shed=repo_elem.find("tool_shed").text,
-                name=repo_elem.find("repository_name").text,
-                owner=repo_elem.find("repository_owner").text,
-                installed_changeset_revision=repo_elem.find("installed_changeset_revision").text,
+                tool_shed=tool_shed_elem.text,
+                name=repository_name_elem.text,
+                owner=repository_owner_elem.text,
+                installed_changeset_revision=installed_changeset_revision_elem.text,
             )
         else:
             repo_info = None
@@ -375,12 +384,13 @@ def configure_and_load(
                     tmp_file.flush()
                 else:
                     # Pull the filename from a global config
-                    filename = file_element.get("from_config", None) or None
+                    filename = file_element.get("from_config")
                     if filename:
                         filename = self.other_config_dict.get(filename, None)
-            filename = file_path = _expand_here_template(filename, here=self.here)
+            if filename:
+                filename = _expand_here_template(filename, here=self.here)
             found = False
-            if file_path is None:
+            if filename is None:
                 log.debug(
                     "Encountered a file element (%s) that does not contain a path value when loading tool data table '%s'.",
                     util.xml_to_string(file_element),
@@ -399,7 +409,7 @@ def configure_and_load(
                 # regular galaxy app has and uses tool_data_path.
                 # We're loading a tool in the tool shed, so we cannot use the Galaxy tool-data
                 # directory which is hard-coded into the tool_data_table_conf.xml entries.
-                filename = os.path.split(file_path)[1]
+                filename = os.path.split(filename)[1]
                 filename = os.path.join(tool_data_path, filename)
             if self.tool_data_path_files.exists(filename):
                 found = True
@@ -526,11 +536,11 @@ def parse_column_spec(self, config_element: Element) -> None:
         else:
             self.largest_index = 0
             for column_elem in config_element.findall("column"):
-                name = column_elem.get("name", None)
+                name = column_elem.get("name")
                 assert name is not None, "Required 'name' attribute missing from column def"
-                index = column_elem.get("index", None)
-                assert index is not None, "Required 'index' attribute missing from column def"
-                index = int(index)
+                index_attr = column_elem.get("index")
+                assert index_attr is not None, "Required 'index' attribute missing from column def"
+                index = int(index_attr)
                 self.columns[name] = index
                 if index > self.largest_index:
                     self.largest_index = index

lib/galaxy/tool_util/data/bundles/models.py

@@ -167,11 +167,11 @@ def _xml_to_data_table_output_column_move(move_elem: Element) -> DataTableBundle
         source_value = ""
     else:
         source_base = source_elem.get("base", None)
-        source_value = source_elem.text
+        source_value = source_elem.text or ""
     target_elem = move_elem.find("target")
     if target_elem is None:
         target_base = None
-        target_value = ""
+        target_value: Optional[str] = ""
     else:
         target_base = target_elem.get("base", None)
         target_value = target_elem.text