More specific RequireAppT type

lib/galaxy/tool_shed/tools/data_table_manager.py

@@ -4,7 +4,6 @@
 from typing import (
     List,
     TYPE_CHECKING,
-    Union,
 )
 
 from galaxy.tool_shed.galaxy_install.client import InstallationTarget
@@ -16,18 +15,48 @@
 from galaxy.util.tool_shed import xml_util
 
 if TYPE_CHECKING:
-    from galaxy.structured_app import BasicSharedApp
+    from galaxy.model.tool_shed_install import ToolShedRepository
+    from tool_shed.structured_app import RequiredAppT
 
 log = logging.getLogger(__name__)
 
 
-RequiredAppT = Union["BasicSharedApp", InstallationTarget]
+class BaseShedToolDataTableManager:
+    def __init__(self, app: "RequiredAppT"):
+        self.app = app
+
+    def handle_sample_tool_data_table_conf_file(self, filename, persist: bool = False):
+        """
+        Parse the incoming filename and add new entries to the in-memory
+        self.app.tool_data_tables dictionary.  If persist is True (should
+        only occur if call is from the Galaxy side, not the tool shed), the
+        new entries will be appended to Galaxy's shed_tool_data_table_conf.xml
+        file on disk.
+        """
+        error = False
+        try:
+            new_table_elems, message = self.app.tool_data_tables.add_new_entries_from_config_file(
+                config_filename=filename,
+                tool_data_path=self.app.config.shed_tool_data_path,
+                shed_tool_data_table_config=self.app.config.shed_tool_data_table_config,
+                persist=persist,
+            )
+            if message:
+                error = True
+        except Exception as e:
+            message = str(e)
+            error = True
+        return error, message
+
+    def reset_tool_data_tables(self):
+        # Reset the tool_data_tables to an empty dictionary.
+        self.app.tool_data_tables.data_tables = {}
 
 
-class ShedToolDataTableManager:
-    app: RequiredAppT
+class ShedToolDataTableManager(BaseShedToolDataTableManager):
+    app: InstallationTarget
 
-    def __init__(self, app: RequiredAppT):
+    def __init__(self, app: InstallationTarget):
         self.app = app
 
     def generate_repository_info_elem(
@@ -105,38 +134,15 @@ def handle_missing_data_table_entry(self, relative_install_dir, tool_path, repos
             self.reset_tool_data_tables()
         return repository_tools_tups
 
-    def handle_sample_tool_data_table_conf_file(self, filename, persist=False):
-        """
-        Parse the incoming filename and add new entries to the in-memory
-        self.app.tool_data_tables dictionary.  If persist is True (should
-        only occur if call is from the Galaxy side, not the tool shed), the
-        new entries will be appended to Galaxy's shed_tool_data_table_conf.xml
-        file on disk.
-        """
-        error = False
-        try:
-            new_table_elems, message = self.app.tool_data_tables.add_new_entries_from_config_file(
-                config_filename=filename,
-                tool_data_path=self.app.config.shed_tool_data_path,
-                shed_tool_data_table_config=self.app.config.shed_tool_data_table_config,
-                persist=persist,
-            )
-            if message:
-                error = True
-        except Exception as e:
-            message = str(e)
-            error = True
-        return error, message
-
-    def get_target_install_dir(self, tool_shed_repository):
+    def get_target_install_dir(self, tool_shed_repository: "ToolShedRepository"):
         tool_path, relative_target_dir = tool_shed_repository.get_tool_relative_path(self.app)
         # This is where index files will reside on a per repo/installed version basis.
         target_dir = os.path.join(self.app.config.shed_tool_data_path, relative_target_dir)
         if not os.path.exists(target_dir):
             os.makedirs(target_dir)
         return target_dir, tool_path, relative_target_dir
 
-    def install_tool_data_tables(self, tool_shed_repository, tool_index_sample_files):
+    def install_tool_data_tables(self, tool_shed_repository: "ToolShedRepository", tool_index_sample_files):
         TOOL_DATA_TABLE_FILE_NAME = "tool_data_table_conf.xml"
         TOOL_DATA_TABLE_FILE_SAMPLE_NAME = f"{TOOL_DATA_TABLE_FILE_NAME}.sample"
         SAMPLE_SUFFIX = ".sample"
@@ -168,7 +174,7 @@ def install_tool_data_tables(self, tool_shed_repository, tool_index_sample_files
             if tree:
                 root = tree.getroot()
                 if root.tag == "tables":
-                    elems = list(root)
+                    elems = list(iter(root))
                 else:
                     log.warning(
                         "The '%s' data table file has '%s' instead of <tables> as root element, skipping.",
@@ -196,10 +202,6 @@ def install_tool_data_tables(self, tool_shed_repository, tool_index_sample_files
             self.app.tool_data_tables.to_xml_file(tool_data_table_conf_filename, elems)
         return tool_data_table_conf_filename, elems
 
-    def reset_tool_data_tables(self):
-        # Reset the tool_data_tables to an empty dictionary.
-        self.app.tool_data_tables.data_tables = {}
-
 
 # For backwards compatibility with exisiting data managers
 ToolDataTableManager = ShedToolDataTableManager

lib/galaxy/tool_shed/tools/tool_validator.py

@@ -1,9 +1,7 @@
 import logging
+from typing import TYPE_CHECKING
 
-from galaxy.tool_shed.tools.data_table_manager import (
-    RequiredAppT,
-    ShedToolDataTableManager,
-)
+from galaxy.tool_shed.tools.data_table_manager import BaseShedToolDataTableManager
 from galaxy.tool_shed.util import (
     basic_util,
     hg_util,
@@ -16,13 +14,16 @@
 )
 from galaxy.tools.parameters import dynamic_options
 
+if TYPE_CHECKING:
+    from tool_shed.structured_app import RequiredAppT
+
 log = logging.getLogger(__name__)
 
 
 class ToolValidator:
-    def __init__(self, app: RequiredAppT):
+    def __init__(self, app: "RequiredAppT"):
         self.app = app
-        self.stdtm = ShedToolDataTableManager(self.app)
+        self.stdtm = BaseShedToolDataTableManager(self.app)
 
     def check_tool_input_params(self, repo_dir, tool_config_name, tool, sample_files):
         """

lib/galaxy/tool_shed/util/repository_util.py

@@ -9,6 +9,7 @@
     List,
     Optional,
     Tuple,
+    TYPE_CHECKING,
     Union,
 )
 from urllib.error import HTTPError
@@ -20,6 +21,7 @@
     or_,
 )
 from sqlalchemy.orm import joinedload
+from typing_extensions import TypeIs
 
 from galaxy import util
 from galaxy.model.base import check_database_connection
@@ -32,6 +34,11 @@
 )
 from galaxy.util.tool_shed.tool_shed_registry import Registry
 
+if TYPE_CHECKING:
+    from galaxy.structured_app import BasicSharedApp
+    from galaxy.tool_shed.galaxy_install.client import InstallationTarget
+    from tool_shed.structured_app import RequiredAppT
+
 log = logging.getLogger(__name__)
 
 VALID_REPOSITORYNAME_RE = re.compile(r"^[a-z0-9\_]+$")
@@ -401,7 +408,7 @@ def get_repository_admin_role_name(repository_name, repository_owner):
     return f"{repository_name}_{repository_owner}_admin"
 
 
-def get_repository_and_repository_dependencies_from_repo_info_dict(app, repo_info_dict):
+def get_repository_and_repository_dependencies_from_repo_info_dict(app: "RequiredAppT", repo_info_dict):
     """Return a tool_shed_repository or repository record defined by the information in the received repo_info_dict."""
     repository_name = list(repo_info_dict.keys())[0]
     repo_info_tuple = repo_info_dict[repository_name]
@@ -414,7 +421,7 @@ def get_repository_and_repository_dependencies_from_repo_info_dict(app, repo_inf
         repository_dependencies,
         tool_dependencies,
     ) = get_repo_info_tuple_contents(repo_info_tuple)
-    if hasattr(app, "install_model"):
+    if is_tool_shed_client(app):
         # In a tool shed client (Galaxy, or something install repositories like Galaxy)
         tool_shed = get_tool_shed_from_clone_url(repository_clone_url)
         repository = get_repository_for_dependency_relationship(
@@ -426,7 +433,7 @@ def get_repository_and_repository_dependencies_from_repo_info_dict(app, repo_inf
     return repository, repository_dependencies
 
 
-def get_repository_by_id(app, id):
+def get_repository_by_id(app: "RequiredAppT", id):
     """Get a repository from the database via id."""
     if is_tool_shed_client(app):
         return app.install_model.context.query(app.install_model.ToolShedRepository).get(app.security.decode_id(id))
@@ -435,7 +442,7 @@ def get_repository_by_id(app, id):
         return sa_session.query(app.model.Repository).get(app.security.decode_id(id))
 
 
-def get_repository_by_name_and_owner(app, name, owner, eagerload_columns=None):
+def get_repository_by_name_and_owner(app: "RequiredAppT", name, owner, eagerload_columns=None):
     """Get a repository from the database via name and owner"""
     repository_query = get_repository_query(app)
     if is_tool_shed_client(app):
@@ -607,15 +614,15 @@ def get_repository_owner_from_clone_url(repository_clone_url):
     return get_repository_owner(tmp_url)
 
 
-def get_repository_query(app):
+def get_repository_query(app: "RequiredAppT"):
     if is_tool_shed_client(app):
         query = app.install_model.context.query(app.install_model.ToolShedRepository)
     else:
         query = app.model.context.query(app.model.Repository)
     return query
 
 
-def get_role_by_id(app, role_id):
+def get_role_by_id(app: "BasicSharedApp", role_id):
     """Get a Role from the database by id."""
     sa_session = app.model.session
     return sa_session.query(app.model.Role).get(app.security.decode_id(role_id))
@@ -680,7 +687,7 @@ def get_tool_shed_status_for_installed_repository(app, repository: ToolShedRepos
     return get_tool_shed_status_for(tool_shed_registry, repository)
 
 
-def is_tool_shed_client(app):
+def is_tool_shed_client(app: "RequiredAppT") -> TypeIs["InstallationTarget"]:
     """
     The tool shed and clients to the tool (i.e. Galaxy) require a lot
     of similar functionality in this file but with small differences. This

lib/tool_shed/structured_app.py

@@ -1,8 +1,13 @@
-from typing import TYPE_CHECKING
+from typing import (
+    TYPE_CHECKING,
+    Union,
+)
 
 from galaxy.structured_app import BasicSharedApp
 
 if TYPE_CHECKING:
+    from galaxy.tool_shed.galaxy_install.client import InstallationTarget
+    from galaxy.tools.data import ToolDataTableManager
     from tool_shed.managers.model_cache import ModelCache
     from tool_shed.repository_registry import RegistryInterface
     from tool_shed.repository_types.registry import Registry as RepositoryTypesRegistry
@@ -18,3 +23,7 @@ class ToolShedApp(BasicSharedApp):
     hgweb_config_manager: "HgWebConfigManager"
     security_agent: "CommunityRBACAgent"
     model_cache: "ModelCache"
+    tool_data_tables: "ToolDataTableManager"
+
+
+RequiredAppT = Union[ToolShedApp, "InstallationTarget"]

lib/tool_shed/util/commit_util.py

@@ -18,9 +18,9 @@
 from sqlalchemy.sql.expression import null
 
 import tool_shed.repository_types.util as rt_util
+from galaxy.tool_shed.tools.data_table_manager import BaseShedToolDataTableManager
 from galaxy.util import checkers
 from galaxy.util.path import safe_relpath
-from tool_shed.tools.data_table_manager import ShedToolDataTableManager
 from tool_shed.util import (
     basic_util,
     hg_util,
@@ -216,7 +216,7 @@ def handle_directory_changes(
             # Handle the special case where a tool_data_table_conf.xml.sample file is being uploaded
             # by parsing the file and adding new entries to the in-memory app.tool_data_tables
             # dictionary.
-            stdtm = ShedToolDataTableManager(app)
+            stdtm = BaseShedToolDataTableManager(app)
             error, message = stdtm.handle_sample_tool_data_table_conf_file(filename_in_archive, persist=False)
             if error:
                 return (

lib/tool_shed/util/repository_util.py

@@ -70,7 +70,10 @@
         ProvidesRepositoriesContext,
         ProvidesUserContext,
     )
-    from tool_shed.structured_app import ToolShedApp
+    from tool_shed.structured_app import (
+        RequiredAppT,
+        ToolShedApp,
+    )
     from tool_shed.webapp.model import Repository
     from tool_shed.webapp.model.mapping import ToolShedModelMapping
 
@@ -250,7 +253,7 @@ def generate_sharable_link_for_repository_in_tool_shed(
     return sharable_url
 
 
-def get_repository_in_tool_shed(app, id, eagerload_columns=None):
+def get_repository_in_tool_shed(app: "RequiredAppT", id, eagerload_columns=None):
     """Get a repository on the tool shed side from the database via id."""
     q = get_repository_query(app)
     if eagerload_columns: