A lot more typing for tool shed install and relevant utility code.

lib/galaxy/structured_app.py

@@ -43,6 +43,7 @@ if TYPE_CHECKING:
     from galaxy.managers.histories import HistoryManager
     from galaxy.managers.workflows import WorkflowsManager
     from galaxy.tools import ToolBox
+    from galaxy.tools.cache import ToolShedRepositoryCache
     from galaxy.tools.data import ToolDataTableManager
     from galaxy.tools.error_reports import ErrorReports
     from galaxy.visualization.genomes import Genomes
@@ -112,6 +113,9 @@ class MinimalManagerApp(MinimalApp):
     def is_job_handler(self) -> bool:
         pass
 
+    def wait_for_toolbox_reload(self, old_toolbox: "ToolBox") -> None:
+        ...
+
 
 class StructuredApp(MinimalManagerApp):
     """Interface defining typed description of the Galaxy UniverseApplication.
@@ -127,6 +131,7 @@ class StructuredApp(MinimalManagerApp):
 
     amqp_internal_connection_obj: Optional[Connection]
     dependency_resolvers_view: DependencyResolversView
+    tool_dependency_dir: Optional[str]
     test_data_resolver: test_data.TestDataResolver
     trs_proxy: TrsProxy
     vault: Vault
@@ -136,7 +141,7 @@ class StructuredApp(MinimalManagerApp):
     tool_data_tables: "ToolDataTableManager"
     tool_cache: Any  # 'galaxy.tools.cache.ToolCache'
     tool_shed_registry: ToolShedRegistry
-    tool_shed_repository_cache: Optional[Any]  # 'galaxy.tools.cache.ToolShedRepositoryCache'
+    tool_shed_repository_cache: Optional["ToolShedRepositoryCache"]
     watchers: "ConfigWatchers"
     workflow_scheduling_manager: Any  # 'galaxy.workflow.scheduling_manager.WorkflowSchedulingManager'
     interactivetool_manager: Any

lib/galaxy/tool_shed/galaxy_install/install_manager.py

@@ -1,6 +1,13 @@
 import json
 import logging
 import os
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+)
 
 from sqlalchemy import or_
 
@@ -8,6 +15,7 @@ from galaxy import (
     exceptions,
     util,
 )
+from galaxy.structured_app import StructuredApp
 from galaxy.tool_shed.galaxy_install.metadata.installed_repository_metadata_manager import (
     InstalledRepositoryMetadataManager,
 )
@@ -33,7 +41,10 @@ log = logging.getLogger(__name__)
 
 
 class InstallRepositoryManager:
-    def __init__(self, app, tpm=None):
+    app: StructuredApp
+    tpm: tool_panel_manager.ToolPanelManager
+
+    def __init__(self, app: StructuredApp, tpm: Optional[tool_panel_manager.ToolPanelManager] = None):
         self.app = app
         self.install_model = self.app.install_model
         self._view = views.DependencyResolversView(app)
@@ -60,7 +71,9 @@ class InstallRepositoryManager:
                 return repo_info_dict, tool_panel_section_key
         return None, None
 
-    def __get_install_info_from_tool_shed(self, tool_shed_url, name, owner, changeset_revision):
+    def __get_install_info_from_tool_shed(
+        self, tool_shed_url: str, name: str, owner: str, changeset_revision: str
+    ) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
         params = dict(name=name, owner=owner, changeset_revision=changeset_revision)
         pathspec = ["api", "repositories", "get_repository_revision_install_info"]
         try:
@@ -302,7 +315,9 @@ class InstallRepositoryManager:
         query = self.install_model.context.query(self.install_model.ToolShedRepository).filter(or_(*clause_list))
         return encoded_kwd, query, tool_shed_repositories, encoded_repository_ids
 
-    def install(self, tool_shed_url, name, owner, changeset_revision, install_options):
+    def install(
+        self, tool_shed_url: str, name: str, owner: str, changeset_revision: str, install_options: Dict[str, Any]
+    ):
         # Get all of the information necessary for installing the repository from the specified tool shed.
         repository_revision_dict, repo_info_dicts = self.__get_install_info_from_tool_shed(
             tool_shed_url, name, owner, changeset_revision
@@ -325,7 +340,11 @@ class InstallRepositoryManager:
         return installed_tool_shed_repositories
 
     def __initiate_and_install_repositories(
-        self, tool_shed_url, repository_revision_dict, repo_info_dicts, install_options
+        self,
+        tool_shed_url: str,
+        repository_revision_dict: Dict[str, Any],
+        repo_info_dicts: List[Dict[str, Any]],
+        install_options: Dict[str, Any],
     ):
         try:
             has_repository_dependencies = repository_revision_dict["has_repository_dependencies"]

lib/galaxy/tool_shed/galaxy_install/installed_repository_manager.py

@@ -5,8 +5,18 @@ import copy
 import logging
 import os
 import shutil
+from typing import (
+    Any,
+    Dict,
+    List,
+    no_type_check,
+    Optional,
+    Tuple,
+)
 
 from galaxy import util
+from galaxy.model.tool_shed_install import ToolShedRepository
+from galaxy.structured_app import MinimalManagerApp
 from galaxy.tool_shed.galaxy_install.metadata.installed_repository_metadata_manager import (
     InstalledRepositoryMetadataManager,
 )
@@ -26,9 +36,18 @@ from galaxy.util.tool_shed.xml_util import parse_xml
 
 log = logging.getLogger(__name__)
 
+RepositoryTupleT = Tuple[str, str, str, str]
+
 
 class InstalledRepositoryManager:
-    def __init__(self, app):
+    app: MinimalManagerApp
+    _tool_paths: List[str]
+    installed_repository_dicts: List[Dict[str, Any]]
+    repository_dependencies_of_installed_repositories: Dict[RepositoryTupleT, List[RepositoryTupleT]]
+    installed_repository_dependencies_of_installed_repositories: Dict[RepositoryTupleT, List[RepositoryTupleT]]
+    installed_dependent_repositories_of_installed_repositories: Dict[RepositoryTupleT, List[RepositoryTupleT]]
+
+    def __init__(self, app: MinimalManagerApp):
         """
         Among other things, keep in in-memory sets of tuples defining installed repositories and tool dependencies along with
         the relationships between each of them.  This will allow for quick discovery of those repositories or components that
@@ -77,7 +96,7 @@ class InstalledRepositoryManager:
             self._tool_paths = tool_paths
         return self._tool_paths
 
-    def activate_repository(self, repository):
+    def activate_repository(self, repository: ToolShedRepository) -> None:
         """Activate an installed tool shed repository that has been marked as deactivated."""
         shed_tool_conf, tool_path, relative_install_dir = suc.get_tool_panel_config_tool_path_install_dir(
             self.app, repository
@@ -123,7 +142,9 @@ class InstalledRepositoryManager:
         self.install_model.session.add(repository)
         self.install_model.session.flush()
 
-    def add_entry_to_installed_repository_dependencies_of_installed_repositories(self, repository):
+    def add_entry_to_installed_repository_dependencies_of_installed_repositories(
+        self, repository: ToolShedRepository
+    ) -> None:
         """
         Add an entry to self.installed_repository_dependencies_of_installed_repositories.  A side-effect of this method
         is the population of self.installed_dependent_repositories_of_installed_repositories.  Since this method discovers
@@ -161,7 +182,7 @@ class InstalledRepositoryManager:
                     repository_tup
                 ]
 
-    def add_entry_to_repository_dependencies_of_installed_repositories(self, repository):
+    def add_entry_to_repository_dependencies_of_installed_repositories(self, repository: ToolShedRepository) -> None:
         """Add an entry to self.repository_dependencies_of_installed_repositories."""
         repository_tup = self.get_repository_tuple_for_installed_repository_manager(repository)
         if repository_tup not in self.repository_dependencies_of_installed_repositories:
@@ -176,25 +197,29 @@ class InstalledRepositoryManager:
             )
             self.repository_dependencies_of_installed_repositories[repository_tup] = repository_dependency_tups
 
-    def get_containing_repository_for_tool_dependency(self, tool_dependency_tup):
+    def get_containing_repository_for_tool_dependency(self, tool_dependency_tup: tuple) -> ToolShedRepository:
         tool_shed_repository_id, name, version, type = tool_dependency_tup
         return self.app.install_model.context.query(self.app.install_model.ToolShedRepository).get(
             tool_shed_repository_id
         )
 
     def get_dependencies_for_repository(
-        self, tool_shed_url, repo_info_dict, includes_tool_dependencies, updating=False
-    ):
+        self,
+        tool_shed_url: str,
+        repo_info_dict: Dict[str, repository_util.AnyRepositoryTupleT],
+        includes_tool_dependencies,
+        updating=False,
+    ) -> Dict[str, Any]:
         """
         Return dictionaries containing the sets of installed and missing tool dependencies and repository
         dependencies associated with the repository defined by the received repo_info_dict.
         """
         rdim = repository_dependency_manager.RepositoryDependencyInstallManager(self.app)
         repository = None
-        installed_rd = {}
-        installed_td = {}
-        missing_rd = {}
-        missing_td = {}
+        installed_rd: Dict[str, Any] = {}
+        installed_td: repository_util.ToolDependenciesDictT = {}
+        missing_rd: Dict[str, Any] = {}
+        missing_td: repository_util.ToolDependenciesDictT = {}
         name = next(iter(repo_info_dict))
         repo_info_tuple = repo_info_dict[name]
         (
@@ -306,7 +331,9 @@ class InstalledRepositoryManager:
         )
         return dependencies_for_repository_dict
 
-    def get_installed_and_missing_repository_dependencies(self, repository):
+    def get_installed_and_missing_repository_dependencies(
+        self, repository: ToolShedRepository
+    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
         """
         Return the installed and missing repository dependencies for a tool shed repository that has a record
         in the Galaxy database, but may or may not be installed.  In this case, the repository dependencies are
@@ -315,8 +342,8 @@ class InstalledRepositoryManager:
         dependencies are really a dependency of the dependent repository's contained tool dependency, and only
         if that tool dependency requires compilation.
         """
-        missing_repository_dependencies = {}
-        installed_repository_dependencies = {}
+        missing_repository_dependencies: Dict[str, Any] = {}
+        installed_repository_dependencies: Dict[str, Any] = {}
         has_repository_dependencies = repository.has_repository_dependencies
         if has_repository_dependencies:
             # The repository dependencies container will include only the immediate repository
@@ -469,7 +496,12 @@ class InstalledRepositoryManager:
             missing_repository_dependencies["description"] = description
         return installed_repository_dependencies, missing_repository_dependencies
 
-    def get_installed_and_missing_tool_dependencies_for_repository(self, tool_dependencies_dict):
+    # The following function will be removed at some point and has clear issues the type checking
+    # makes clear... I'm going to skip type checking for now rather than fix bugs in deprecated code
+    @no_type_check
+    def get_installed_and_missing_tool_dependencies_for_repository(
+        self, tool_dependencies_dict: repository_util.ToolDependenciesDictT
+    ) -> Tuple[repository_util.ToolDependenciesDictT, repository_util.ToolDependenciesDictT]:
         """
         Return the lists of installed tool dependencies and missing tool dependencies for a set of repositories
         being installed into Galaxy.
@@ -480,16 +512,17 @@ class InstalledRepositoryManager:
         # package repository approach).  If 2 repositories contain the same tool_dependencies.xml file, one dependency
         # will be lost since the values in these returned dictionaries are not lists.  All tool dependency dictionaries
         # should have lists as values.  These scenarios are probably extreme corner cases, but still should be handled.
-        installed_tool_dependencies = {}
-        missing_tool_dependencies = {}
+        installed_tool_dependencies: dict = {}
+        missing_tool_dependencies: dict = {}
         if tool_dependencies_dict:
             # Make sure not to change anything in the received tool_dependencies_dict as that would be a bad side-effect!
-            tmp_tool_dependencies_dict = copy.deepcopy(tool_dependencies_dict)
+            tmp_tool_dependencies_dict: dict = copy.deepcopy(tool_dependencies_dict)
             for td_key, val in tmp_tool_dependencies_dict.items():
                 # Default the status to NEVER_INSTALLED.
                 tool_dependency_status = self.install_model.ToolDependency.installation_status.NEVER_INSTALLED
-                # Set environment tool dependencies are a list.
                 if td_key == "set_environment":
+                    # Set environment tool dependencies are a list.
+                    assert isinstance(val, list)
                     new_val = []
                     for requirement_dict in val:
                         # {'repository_name': 'xx',
@@ -564,7 +597,7 @@ class InstalledRepositoryManager:
                     )
         return dependency_tups
 
-    def get_repository_tuple_for_installed_repository_manager(self, repository):
+    def get_repository_tuple_for_installed_repository_manager(self, repository: ToolShedRepository) -> RepositoryTupleT:
         return (
             str(repository.tool_shed),
             str(repository.name),
@@ -572,7 +605,7 @@ class InstalledRepositoryManager:
             str(repository.installed_changeset_revision),
         )
 
-    def get_repository_install_dir(self, tool_shed_repository):
+    def get_repository_install_dir(self, tool_shed_repository: ToolShedRepository) -> Optional[str]:
         for tool_path in self.tool_paths:
             ts = common_util.remove_port_from_tool_shed_url(str(tool_shed_repository.tool_shed))
             relative_path = os.path.join(
@@ -588,8 +621,8 @@ class InstalledRepositoryManager:
         return None
 
     def handle_existing_tool_dependencies_that_changed_in_update(
-        self, repository, original_dependency_dict, new_dependency_dict
-    ):
+        self, repository: ToolShedRepository, original_dependency_dict, new_dependency_dict
+    ) -> Tuple[List[str], List[str]]:
         """
         This method is called when a Galaxy admin is getting updates for an installed tool shed
         repository in order to cover the case where an existing tool dependency was changed (e.g.,
@@ -603,7 +636,7 @@ class InstalledRepositoryManager:
         deleted_tool_dependency_names = []
         for original_dependency_key, original_dependency_val_dict in original_dependency_dict.items():
             if original_dependency_key not in new_dependency_dict:
-                updated_tool_dependency = self.update_existing_tool_dependency(
+                updated_tool_dependency = self._update_existing_tool_dependency(
                     repository, original_dependency_val_dict, new_dependency_dict
                 )
                 if updated_tool_dependency:
@@ -612,7 +645,7 @@ class InstalledRepositoryManager:
                     deleted_tool_dependency_names.append(original_dependency_val_dict["name"])
         return updated_tool_dependency_names, deleted_tool_dependency_names
 
-    def uninstall_repository(self, repository, remove_from_disk=True):
+    def uninstall_repository(self, repository: ToolShedRepository, remove_from_disk=True):
         errors = ""
         shed_tool_conf, tool_path, relative_install_dir = suc.get_tool_panel_config_tool_path_install_dir(
             app=self.app, repository=repository
@@ -663,7 +696,7 @@ class InstalledRepositoryManager:
         self.app.install_model.session.flush()
         return errors
 
-    def purge_repository(self, repository):
+    def purge_repository(self, repository: ToolShedRepository) -> Tuple[str, str]:
         """Purge a repository with status New (a white ghost) from the database."""
         sa_session = self.app.model.session
         status = "ok"
@@ -800,7 +833,9 @@ class InstalledRepositoryManager:
             message += f" the status {repository.status}."
         return status, message
 
-    def remove_entry_from_installed_repository_dependencies_of_installed_repositories(self, repository):
+    def remove_entry_from_installed_repository_dependencies_of_installed_repositories(
+        self, repository: ToolShedRepository
+    ):
         """
         Remove an entry from self.installed_repository_dependencies_of_installed_repositories.  A side-effect of this method
         is removal of appropriate value items from self.installed_dependent_repositories_of_installed_repositories.
@@ -832,7 +867,9 @@ class InstalledRepositoryManager:
             log.debug(debug_msg)
             del self.installed_repository_dependencies_of_installed_repositories[repository_tup]
 
-    def remove_entry_from_repository_dependencies_of_installed_repositories(self, repository):
+    def remove_entry_from_repository_dependencies_of_installed_repositories(
+        self, repository: ToolShedRepository
+    ) -> None:
         """Remove an entry from self.repository_dependencies_of_installed_repositories."""
         repository_tup = self.get_repository_tuple_for_installed_repository_manager(repository)
         if repository_tup in self.repository_dependencies_of_installed_repositories:
@@ -844,7 +881,9 @@ class InstalledRepositoryManager:
             log.debug(debug_msg)
             del self.repository_dependencies_of_installed_repositories[repository_tup]
 
-    def repository_dependency_needed_only_for_compiling_tool_dependency(self, repository, repository_dependency):
+    def repository_dependency_needed_only_for_compiling_tool_dependency(
+        self, repository: ToolShedRepository, repository_dependency
+    ) -> bool:
         for rd_tup in repository.tuples_of_repository_dependencies_needed_for_compiling_td:
             (
                 tool_shed,
@@ -918,7 +957,9 @@ class InstalledRepositoryManager:
                 return str(required_rd_tup[4])
         return "False"
 
-    def update_existing_tool_dependency(self, repository, original_dependency_dict, new_dependencies_dict):
+    def _update_existing_tool_dependency(
+        self, repository: ToolShedRepository, original_dependency_dict, new_dependencies_dict
+    ):
         """
         Update an exsiting tool dependency whose definition was updated in a change set
         pulled by a Galaxy administrator when getting updates to an installed tool shed

lib/galaxy/tool_shed/galaxy_install/metadata/installed_repository_metadata_manager.py

@@ -1,9 +1,11 @@
 import logging
 import os
+from typing import Optional
 
 from sqlalchemy import false
 
 from galaxy import util
+from galaxy.structured_app import MinimalManagerApp
 from galaxy.tool_shed.galaxy_install.tools import tool_panel_manager
 from galaxy.tool_shed.metadata.metadata_generator import MetadataGenerator
 from galaxy.tool_shed.util.repository_util import (
@@ -24,8 +26,8 @@ log = logging.getLogger(__name__)
 class InstalledRepositoryMetadataManager(MetadataGenerator):
     def __init__(
         self,
-        app,
-        tpm=None,
+        app: MinimalManagerApp,
+        tpm: Optional[tool_panel_manager.ToolPanelManager] = None,
         repository=None,
         changeset_revision=None,
         repository_clone_url=None,

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

@@ -2,16 +2,28 @@ import errno
 import logging
 import os
 import time
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    TYPE_CHECKING,
+)
 
 from galaxy.util import (
+    Element,
     etree,
     parse_xml_string,
     xml_to_string,
 )
+from galaxy.util.path import StrPath
 from galaxy.util.renamed_temporary_file import RenamedTemporaryFile
 from galaxy.util.tool_shed.xml_util import parse_xml
 from . import tool_panel_manager
 
+if TYPE_CHECKING:
+    from galaxy.tools.data_manager.manager import DataManager
+
 log = logging.getLogger(__name__)
 
 SHED_DATA_MANAGER_CONF_XML = """<?xml version="1.0"?>
@@ -21,18 +33,20 @@ SHED_DATA_MANAGER_CONF_XML = """<?xml version="1.0"?>
 
 
 class DataManagerHandler:
+    root: Optional[Element] = None
+
     def __init__(self, app):
         self.app = app
 
     @property
-    def data_managers_path(self):
+    def data_managers_path(self) -> Optional[str]:
         tree, error_message = parse_xml(self.app.config.shed_data_manager_config_file)
         if tree:
             root = tree.getroot()
             return root.get("tool_path", None)
         return None
 
-    def data_manager_config_elems_to_xml_file(self, config_elems, config_filename):
+    def _data_manager_config_elems_to_xml_file(self, config_elems: List[Element], config_filename: StrPath) -> None:
         """
         Persist the current in-memory list of config_elems to a file named by the value
         of config_filename.
@@ -49,18 +63,18 @@ class DataManagerHandler:
             with RenamedTemporaryFile(config_filename, mode="w") as fh:
                 fh.write(xml_to_string(root))
         except Exception:
-            log.exception("Exception in DataManagerHandler.data_manager_config_elems_to_xml_file")
+            log.exception("Exception in DataManagerHandler._data_manager_config_elems_to_xml_file")
 
     def install_data_managers(
         self,
-        shed_data_manager_conf_filename,
-        metadata_dict,
-        shed_config_dict,
-        relative_install_dir,
+        shed_data_manager_conf_filename: StrPath,
+        metadata_dict: Dict[str, Any],
+        shed_config_dict: Dict[str, Any],
+        relative_install_dir: StrPath,
         repository,
         repository_tools_tups,
     ):
-        rval = []
+        rval: List["DataManager"] = []
         if "data_manager" in metadata_dict:
             tpm = tool_panel_manager.ToolPanelManager(self.app)
             repository_tools_by_guid = {}
@@ -163,7 +177,7 @@ class DataManagerHandler:
             # Persist the altered shed_data_manager_config file.
             if data_manager_config_has_changes:
                 reload_count = self.app.data_managers._reload_count
-                self.data_manager_config_elems_to_xml_file(config_elems, shed_data_manager_conf_filename)
+                self._data_manager_config_elems_to_xml_file(config_elems, shed_data_manager_conf_filename)
                 while self.app.data_managers._reload_count <= reload_count:
                     time.sleep(0.1)  # Wait for shed_data_manager watcher thread to pick up changes
         return rval
@@ -215,4 +229,4 @@ class DataManagerHandler:
                     self.app.data_managers.load_manager_from_elem(elem)
                 # Persist the altered shed_data_manager_config file.
                 if data_manager_config_has_changes:
-                    self.data_manager_config_elems_to_xml_file(config_elems, shed_data_manager_conf_filename)
+                    self._data_manager_config_elems_to_xml_file(config_elems, shed_data_manager_conf_filename)

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

@@ -1,7 +1,13 @@
 import errno
 import logging
+from typing import (
+    Any,
+    Dict,
+    List,
+)
 
 from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.structured_app import MinimalManagerApp
 from galaxy.tool_shed.util.basic_util import strip_path
 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
@@ -18,10 +24,12 @@ log = logging.getLogger(__name__)
 
 
 class ToolPanelManager:
-    def __init__(self, app):
+    app: MinimalManagerApp
+
+    def __init__(self, app: MinimalManagerApp):
         self.app = app
 
-    def add_to_shed_tool_config(self, shed_tool_conf_dict, elem_list):
+    def add_to_shed_tool_config(self, shed_tool_conf_dict: Dict[str, Any], elem_list: list) -> None:
         """
         "A tool shed repository is being installed so change the shed_tool_conf file.  Parse the
         config file to generate the entire list of config_elems instead of using the in-memory list
@@ -175,7 +183,7 @@ class ToolPanelManager:
         currently be defined within the same tool section in the tool panel or
         outside of any sections.
         """
-        tool_panel_dict = {}
+        tool_panel_dict: Dict[str, List[Dict[str, Any]]] = {}
         if tool_section:
             section_id = tool_section.id
             section_name = tool_section.name
@@ -197,7 +205,9 @@ class ToolPanelManager:
                     tool_panel_dict[guid] = [tool_section_dict]
         return tool_panel_dict
 
-    def generate_tool_panel_dict_for_tool_config(self, guid, tool_config, tool_sections=None):
+    def generate_tool_panel_dict_for_tool_config(
+        self, guid, tool_config, tool_sections=None
+    ) -> Dict[str, List[Dict[str, Any]]]:
         """
         Create a dictionary of the following type for a single tool config file name.
         The intent is to call this method for every tool config in a repository and
@@ -213,13 +223,13 @@ class ToolPanelManager:
                     name : <TooSection name>}]}
 
         """
-        tool_panel_dict = {}
+        tool_panel_dict: Dict[str, List[Dict[str, Any]]] = {}
         file_name = strip_path(tool_config)
         tool_section_dicts = self.generate_tool_section_dicts(tool_config=file_name, tool_sections=tool_sections)
         tool_panel_dict[guid] = tool_section_dicts
         return tool_panel_dict
 
-    def generate_tool_panel_dict_from_shed_tool_conf_entries(self, repository):
+    def generate_tool_panel_dict_from_shed_tool_conf_entries(self, repository) -> Dict[str, List[Dict[str, Any]]]:
         """
         Keep track of the section in the tool panel in which this repository's
         tools will be contained by parsing the shed_tool_conf in which the
@@ -228,7 +238,7 @@ class ToolPanelManager:
         repository is being deactivated or un-installed and allows for
         activation or re-installation using the original layout.
         """
-        tool_panel_dict = {}
+        tool_panel_dict: Dict[str, List[Dict[str, Any]]] = {}
         shed_tool_conf, tool_path, relative_install_dir = get_tool_panel_config_tool_path_install_dir(
             self.app, repository
         )
@@ -279,15 +289,15 @@ class ToolPanelManager:
 
     def generate_tool_panel_elem_list(
         self,
-        repository_name,
-        repository_clone_url,
-        changeset_revision,
-        tool_panel_dict,
-        repository_tools_tups,
+        repository_name: str,
+        repository_clone_url: str,
+        changeset_revision: str,
+        tool_panel_dict: dict,
+        repository_tools_tups: List[tuple],
         owner="",
     ):
         """Generate a list of ElementTree Element objects for each section or tool."""
-        elem_list = []
+        elem_list: List[etree.Element] = []
         tool_elem = None
         cleaned_repository_clone_url = remove_protocol_and_user_from_clone_url(repository_clone_url)
         if not owner:
@@ -333,8 +343,8 @@ class ToolPanelManager:
                     elem_list.append(tool_elem)
         return elem_list
 
-    def generate_tool_section_dicts(self, tool_config=None, tool_sections=None):
-        tool_section_dicts = []
+    def generate_tool_section_dicts(self, tool_config=None, tool_sections=None) -> List[Dict[str, Any]]:
+        tool_section_dicts: List[Dict[str, Any]] = []
         if tool_config is None:
             tool_config = ""
         if tool_sections:

lib/galaxy/tool_shed/tools/data_table_manager.py

@@ -1,7 +1,9 @@
 import logging
 import os
 import shutil
+from typing import List
 
+from galaxy.structured_app import StructuredApp
 from galaxy.tool_shed.util import hg_util
 from galaxy.util import etree
 from galaxy.util.tool_shed import xml_util
@@ -10,12 +12,12 @@ log = logging.getLogger(__name__)
 
 
 class ShedToolDataTableManager:
-    def __init__(self, app):
+    def __init__(self, app: StructuredApp):
         self.app = app
 
     def generate_repository_info_elem(
-        self, tool_shed, repository_name, changeset_revision, owner, parent_elem=None, **kwd
-    ):
+        self, tool_shed: str, repository_name: str, changeset_revision: str, owner: str, parent_elem=None, **kwd
+    ) -> etree.Element:
         """Create and return an ElementTree repository info Element."""
         if parent_elem is None:
             elem = etree.Element("tool_shed_repository")
@@ -46,7 +48,7 @@ class ShedToolDataTableManager:
             **kwd,
         )
 
-    def get_tool_index_sample_files(self, sample_files):
+    def get_tool_index_sample_files(self, sample_files: List[str]) -> List[str]:
         """
         Try to return the list of all appropriate tool data sample files included
         in the repository.

lib/galaxy/tool_shed/util/container_util.py

@@ -9,13 +9,13 @@ STRSEP = "__ESEP__"
 
 
 def generate_repository_dependencies_key_for_repository(
-    toolshed_base_url,
-    repository_name,
-    repository_owner,
-    changeset_revision,
-    prior_installation_required,
-    only_if_compiling_contained_td,
-):
+    toolshed_base_url: str,
+    repository_name: str,
+    repository_owner: str,
+    changeset_revision: str,
+    prior_installation_required: bool,
+    only_if_compiling_contained_td: bool,
+) -> str:
     """
     Assumes tool shed is current tool shed since repository dependencies across tool sheds
     is not yet supported.
@@ -39,7 +39,7 @@ def generate_repository_dependencies_key_for_repository(
     )
 
 
-def get_components_from_key(key):
+def get_components_from_key(key: str) -> tuple:
     """
     Assumes tool shed is current tool shed since repository dependencies across tool sheds is not
     yet supported.

lib/galaxy/tool_shed/util/hg_util.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import subprocess
+from typing import Optional
 
 from galaxy.tool_shed.util import basic_util
 from galaxy.util import unicodify
@@ -59,7 +60,7 @@ def get_changectx_for_changeset(repo, changeset_revision, **kwd):
     return None
 
 
-def get_config_from_disk(config_file, relative_install_dir):
+def get_config_from_disk(config_file: str, relative_install_dir: str) -> Optional[str]:
     for root, _dirs, files in os.walk(relative_install_dir):
         if root.find(".hg") < 0:
             for name in files:

lib/galaxy/tool_shed/util/repository_util.py

@@ -2,6 +2,15 @@ import logging
 import os
 import re
 import shutil
+from typing import (
+    Any,
+    cast,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
 from urllib.error import HTTPError
 
 from markupsafe import escape
@@ -332,9 +341,16 @@ def get_prior_import_or_install_required_dict(app, tsr_ids, repo_info_dicts):
     return prior_import_or_install_required_dict
 
 
-def get_repo_info_tuple_contents(repo_info_tuple):
+ToolDependenciesDictT = Dict[str, Union[Dict[str, Any], List[Dict[str, Any]]]]
+OldRepositoryTupleT = Tuple[str, str, str, str, str, ToolDependenciesDictT]
+RepositoryTupleT = Tuple[str, str, str, str, str, Optional[Any], ToolDependenciesDictT]
+AnyRepositoryTupleT = Union[OldRepositoryTupleT, RepositoryTupleT]
+
+
+def get_repo_info_tuple_contents(repo_info_tuple: AnyRepositoryTupleT) -> RepositoryTupleT:
     """Take care in handling the repo_info_tuple as it evolves over time as new tool shed features are introduced."""
     if len(repo_info_tuple) == 6:
+        old_repo_info = cast(OldRepositoryTupleT, repo_info_tuple)
         (
             description,
             repository_clone_url,
@@ -342,9 +358,10 @@ def get_repo_info_tuple_contents(repo_info_tuple):
             ctx_rev,
             repository_owner,
             tool_dependencies,
-        ) = repo_info_tuple
+        ) = old_repo_info
         repository_dependencies = None
     elif len(repo_info_tuple) == 7:
+        repo_info = cast(RepositoryTupleT, repo_info_tuple)
         (
             description,
             repository_clone_url,
@@ -353,7 +370,7 @@ def get_repo_info_tuple_contents(repo_info_tuple):
             repository_owner,
             repository_dependencies,
             tool_dependencies,
-        ) = repo_info_tuple
+        ) = repo_info
     return (
         description,
         repository_clone_url,

lib/galaxy/tools/data_manager/manager.py

@@ -2,7 +2,10 @@ import errno
 import json
 import logging
 import os
-from typing import Dict
+from typing import (
+    Dict,
+    Optional,
+)
 
 from galaxy import util
 from galaxy.structured_app import MinimalManagerApp
@@ -38,7 +41,7 @@ class DataManagers:
                 if exc.errno != errno.ENOENT or self.app.config.is_set("shed_data_manager_config_file"):
                     raise
 
-    def load_from_xml(self, xml_filename, store_tool_path=True):
+    def load_from_xml(self, xml_filename, store_tool_path=True) -> None:
         try:
             tree = util.parse_xml(xml_filename)
         except OSError as e:
@@ -68,7 +71,7 @@ class DataManagers:
                 tool_path = os.path.dirname(xml_filename)
                 self.load_manager_from_elem(data_manager_elem, tool_path=tool_path)
 
-    def load_manager_from_elem(self, data_manager_elem, tool_path=None, add_manager=True):
+    def load_manager_from_elem(self, data_manager_elem, tool_path=None, add_manager=True) -> Optional["DataManager"]:
         try:
             data_manager = DataManager(self, data_manager_elem, tool_path=tool_path)
         except OSError as e:

lib/galaxy/util/__init__.py

@@ -85,6 +85,7 @@ from .path import (  # noqa: F401
     safe_contains,
     safe_makedirs,
     safe_relpath,
+    StrPath,
 )
 
 try:
@@ -283,7 +284,7 @@ def unique_id(KEY_SIZE=128):
     return md5(random_bits).hexdigest()
 
 
-def parse_xml(fname: typing.Union[str, os.PathLike], strip_whitespace=True, remove_comments=True):
+def parse_xml(fname: StrPath, strip_whitespace=True, remove_comments=True) -> etree.ElementTree:
     """Returns a parsed xml tree"""
     parser = None
     if remove_comments and LXML_AVAILABLE:
@@ -310,7 +311,7 @@ def parse_xml(fname: typing.Union[str, os.PathLike], strip_whitespace=True, remo
     return tree
 
 
-def parse_xml_string(xml_string, strip_whitespace=True):
+def parse_xml_string(xml_string, strip_whitespace=True) -> etree.Element:
     try:
         tree = etree.fromstring(xml_string)
     except ValueError as e:

lib/galaxy/util/renamed_temporary_file.py

@@ -3,6 +3,8 @@
 import os
 import tempfile
 
+from galaxy.util.path import StrOrBytesPath
+
 
 class RenamedTemporaryFile:
     """
@@ -10,7 +12,9 @@ class RenamedTemporaryFile:
     path on exit.
     """
 
-    def __init__(self, final_path, **kwargs):
+    final_path: StrOrBytesPath
+
+    def __init__(self, final_path: StrOrBytesPath, **kwargs):
         """
         >>> dir = tempfile.mkdtemp()
         >>> with RenamedTemporaryFile(os.path.join(dir, 'test.txt'), mode="w") as out:
@@ -27,7 +31,7 @@ class RenamedTemporaryFile:
         self.tmpfile = tempfile.NamedTemporaryFile(dir=tmpfile_dir, delete=False, **kwargs)
         self.final_path = final_path
 
-    def __getattr__(self, attr):
+    def __getattr__(self, attr: str):
         """
         Delegate attribute access to the underlying temporary file object.
         """

lib/galaxy/util/tool_shed/common_util.py

@@ -1,18 +1,31 @@
 import json
 import logging
 import os
+from typing import (
+    Optional,
+    TYPE_CHECKING,
+)
 from urllib.parse import urljoin
 
 from routes import url_for
+from typing_extensions import Protocol
 
 from galaxy import util
 from galaxy.util.tool_shed import encoding_util
 
+if TYPE_CHECKING:
+    from .tool_shed_registry import Registry as ToolShedRegistry
+
 log = logging.getLogger(__name__)
 
 REPOSITORY_OWNER = "devteam"
 
 
+class HasToolShedRegistry(Protocol):
+    tool_shed_registry: "ToolShedRegistry"
+    name: str
+
+
 def accumulate_tool_dependencies(tool_shed_accessible, tool_dependencies, all_tool_dependencies):
     if tool_shed_accessible:
         if tool_dependencies:
@@ -32,13 +45,13 @@ def check_tool_tag_set(elem, migrated_tool_configs_dict, missing_tool_configs_di
     return missing_tool_configs_dict
 
 
-def generate_clone_url_for_installed_repository(app, repository):
+def generate_clone_url_for_installed_repository(app: HasToolShedRegistry, repository) -> str:
     """Generate the URL for cloning a repository that has been installed into a Galaxy instance."""
     tool_shed_url = get_tool_shed_url_from_tool_shed_registry(app, str(repository.tool_shed))
     return util.build_url(tool_shed_url, pathspec=["repos", str(repository.owner), str(repository.name)])
 
 
-def generate_clone_url_for_repository_in_tool_shed(user, repository):
+def generate_clone_url_for_repository_in_tool_shed(user, repository) -> str:
     """Generate the URL for cloning a repository that is in the tool shed."""
     base_url = url_for("/", qualified=True).rstrip("/")
     if user:
@@ -49,7 +62,7 @@ def generate_clone_url_for_repository_in_tool_shed(user, repository):
         return f"{base_url}/repos/{repository.user.username}/{repository.name}"
 
 
-def generate_clone_url_from_repo_info_tup(app, repo_info_tup):
+def generate_clone_url_from_repo_info_tup(app: HasToolShedRegistry, repo_info_tup) -> str:
     """Generate the URL for cloning a repository given a tuple of toolshed, name, owner, changeset_revision."""
     # Example tuple: ['http://localhost:9009', 'blast_datatypes', 'test', '461a4216e8ab', False]
     (
@@ -89,7 +102,7 @@ def get_repository_dependencies(app, tool_shed_url, repository_name, repository_
     return tool_shed_accessible, repository_dependencies_dict
 
 
-def get_protocol_from_tool_shed_url(tool_shed_url):
+def get_protocol_from_tool_shed_url(tool_shed_url: str) -> str:
     """Return the protocol from the received tool_shed_url if it exists."""
     try:
         if tool_shed_url.find("://") > 0:
@@ -99,8 +112,8 @@ def get_protocol_from_tool_shed_url(tool_shed_url):
         # that value when creating a header row.  If the tool_shed_url is not None, we have a problem.
         if tool_shed_url is not None:
             log.exception("Handled exception getting the protocol from Tool Shed URL %s", str(tool_shed_url))
-        # Default to HTTP protocol.
-        return "http"
+    # Default to HTTP protocol.
+    return "http"
 
 
 def get_tool_shed_repository_ids(as_string=False, **kwd):
@@ -125,7 +138,7 @@ def get_tool_shed_repository_ids(as_string=False, **kwd):
     return []
 
 
-def get_tool_shed_url_from_tool_shed_registry(app, tool_shed):
+def get_tool_shed_url_from_tool_shed_registry(app: HasToolShedRegistry, tool_shed: str) -> Optional[str]:
     """
     The value of tool_shed is something like: toolshed.g2.bx.psu.edu.  We need the URL to this tool shed, which is
     something like: http://toolshed.g2.bx.psu.edu/
@@ -140,7 +153,7 @@ def get_tool_shed_url_from_tool_shed_registry(app, tool_shed):
     return None
 
 
-def get_tool_shed_repository_url(app, tool_shed, owner, name):
+def get_tool_shed_repository_url(app: HasToolShedRegistry, tool_shed: str, owner: str, name: str):
     tool_shed_url = get_tool_shed_url_from_tool_shed_registry(app, tool_shed)
     if tool_shed_url:
         # Append a slash to the tool shed URL, because urlparse.urljoin will eliminate
@@ -169,12 +182,13 @@ def handle_galaxy_url(trans, **kwd):
     return galaxy_url
 
 
-def handle_tool_shed_url_protocol(app, shed_url):
+def handle_tool_shed_url_protocol(app: HasToolShedRegistry, shed_url: str) -> str:
     """Handle secure and insecure HTTP protocol since they may change over time."""
     try:
         if app.name == "galaxy":
             url = remove_protocol_from_tool_shed_url(shed_url)
             tool_shed_url = get_tool_shed_url_from_tool_shed_registry(app, url)
+            assert tool_shed_url
         else:
             tool_shed_url = str(url_for("/", qualified=True)).rstrip("/")
         return tool_shed_url
@@ -232,7 +246,7 @@ def parse_repository_dependency_tuple(repository_dependency_tuple, contains_erro
         return tool_shed, name, owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td
 
 
-def remove_port_from_tool_shed_url(tool_shed_url):
+def remove_port_from_tool_shed_url(tool_shed_url: str) -> str:
     """Return a partial Tool Shed URL, eliminating the port if it exists."""
     try:
         if tool_shed_url.find(":") > 0:
@@ -249,14 +263,14 @@ def remove_port_from_tool_shed_url(tool_shed_url):
         return tool_shed_url
 
 
-def remove_protocol_and_port_from_tool_shed_url(tool_shed_url):
+def remove_protocol_and_port_from_tool_shed_url(tool_shed_url: str) -> str:
     """Return a partial Tool Shed URL, eliminating the protocol and/or port if either exists."""
     tool_shed = remove_protocol_from_tool_shed_url(tool_shed_url)
     tool_shed = remove_port_from_tool_shed_url(tool_shed)
     return tool_shed
 
 
-def remove_protocol_and_user_from_clone_url(repository_clone_url):
+def remove_protocol_and_user_from_clone_url(repository_clone_url: str) -> str:
     """Return a URL that can be used to clone a repository, eliminating the protocol and user if either exists."""
     if repository_clone_url.find("@") > 0:
         # We have an url that includes an authenticated user, something like:
@@ -273,7 +287,7 @@ def remove_protocol_and_user_from_clone_url(repository_clone_url):
     return tmp_url.rstrip("/")
 
 
-def remove_protocol_from_tool_shed_url(tool_shed_url):
+def remove_protocol_from_tool_shed_url(tool_shed_url: str) -> str:
     """Return a partial Tool Shed URL, eliminating the protocol if it exists."""
     return util.remove_protocol_from_url(tool_shed_url)
 

lib/galaxy/util/tool_shed/xml_util.py

@@ -17,7 +17,7 @@ from galaxy.util.path import StrPath
 log = logging.getLogger(__name__)
 
 
-def create_and_write_tmp_file(elem):
+def create_and_write_tmp_file(elem: etree.Element) -> str:
     tmp_str = xml_to_string(elem, pretty=True)
     with tempfile.NamedTemporaryFile(prefix="tmp-toolshed-cawrf", delete=False) as fh:
         tmp_filename = fh.name

mypy.ini

@@ -382,8 +382,6 @@ check_untyped_defs = False
 check_untyped_defs = False
 [mypy-galaxy.tool_util.deps.resolvers.conda]
 check_untyped_defs = False
-[mypy-galaxy.tool_shed.galaxy_install.tools.tool_panel_manager]
-check_untyped_defs = False
 [mypy-galaxy.jobs.rule_helper]
 check_untyped_defs = False
 [mypy-galaxy.datatypes.isa]
@@ -396,8 +394,6 @@ check_untyped_defs = False
 check_untyped_defs = False
 [mypy-galaxy.tool_util.parser.cwl]
 check_untyped_defs = False
-[mypy-galaxy.tool_shed.galaxy_install.tools.data_manager]
-check_untyped_defs = False
 [mypy-galaxy.tool_shed.galaxy_install.repository_dependencies.repository_dependency_manager]
 check_untyped_defs = False
 [mypy-galaxy.datatypes.triples]