From db66a44ba7e33126040b25c685a0eca4d8b537b3 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Wed, 12 Feb 2025 10:33:25 +0000 Subject: [PATCH] More specific ``RequireAppT`` type --- .../tool_shed/tools/data_table_manager.py | 74 ++++++++++--------- lib/galaxy/tool_shed/tools/tool_validator.py | 13 ++-- lib/galaxy/tool_shed/util/repository_util.py | 21 ++++-- lib/tool_shed/structured_app.py | 11 ++- lib/tool_shed/util/commit_util.py | 4 +- lib/tool_shed/util/repository_util.py | 7 +- 6 files changed, 76 insertions(+), 54 deletions(-) diff --git a/lib/galaxy/tool_shed/tools/data_table_manager.py b/lib/galaxy/tool_shed/tools/data_table_manager.py index f5d0a5669a7d..77beb6f7dcb0 100644 --- a/lib/galaxy/tool_shed/tools/data_table_manager.py +++ b/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,30 +134,7 @@ 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) @@ -136,7 +142,7 @@ def get_target_install_dir(self, tool_shed_repository): 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 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 diff --git a/lib/galaxy/tool_shed/tools/tool_validator.py b/lib/galaxy/tool_shed/tools/tool_validator.py index 428601c65113..9f42cf009963 100644 --- a/lib/galaxy/tool_shed/tools/tool_validator.py +++ b/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): """ diff --git a/lib/galaxy/tool_shed/util/repository_util.py b/lib/galaxy/tool_shed/util/repository_util.py index 5f9a4cd0ff7b..d575088869e1 100644 --- a/lib/galaxy/tool_shed/util/repository_util.py +++ b/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,7 +614,7 @@ 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: @@ -615,7 +622,7 @@ def get_repository_query(app): 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 diff --git a/lib/tool_shed/structured_app.py b/lib/tool_shed/structured_app.py index 8a38e008fbbe..e384682223fd 100644 --- a/lib/tool_shed/structured_app.py +++ b/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"] diff --git a/lib/tool_shed/util/commit_util.py b/lib/tool_shed/util/commit_util.py index ab3864f8e98d..afbaa3f88039 100644 --- a/lib/tool_shed/util/commit_util.py +++ b/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 ( diff --git a/lib/tool_shed/util/repository_util.py b/lib/tool_shed/util/repository_util.py index 51be7f9d5535..66fcdb837e6c 100644 --- a/lib/tool_shed/util/repository_util.py +++ b/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: