diff --git a/lib/galaxy/managers/tool_data.py b/lib/galaxy/managers/tool_data.py index 03f0caa9bbc..170f3f61144 100644 --- a/lib/galaxy/managers/tool_data.py +++ b/lib/galaxy/managers/tool_data.py @@ -1,5 +1,6 @@ from pathlib import Path from typing import ( + cast, Dict, Optional, ) @@ -9,6 +10,7 @@ from galaxy.structured_app import StructuredApp from galaxy.tools.data import ( TabularToolDataField, TabularToolDataTable, + ToolDataTable, ) from galaxy.tools.data._schema import ( ToolDataDetails, @@ -26,7 +28,7 @@ class ToolDataManager: self._app = app @property - def data_tables(self) -> Dict[str, TabularToolDataTable]: + def data_tables(self) -> Dict[str, ToolDataTable]: return self._app.tool_data_tables.data_tables def index(self) -> ToolDataEntryList: @@ -61,7 +63,7 @@ class ToolDataManager: def delete(self, table_name: str, values: Optional[str] = None) -> ToolDataDetails: """Removes an item from a data table""" - data_table = self._data_table(table_name) + data_table = self._tabular_data_table(table_name) if not values: raise exceptions.RequestParameterInvalidException("Invalid values for data table item specified.") @@ -75,14 +77,17 @@ class ToolDataManager: data_table.remove_entry(split_values) return self._reload_data_table(table_name) - def _data_table(self, table_name: str) -> TabularToolDataTable: + def _tabular_data_table(self, table_name: str) -> TabularToolDataTable: + return cast(TabularToolDataTable, self._data_table(table_name)) + + def _data_table(self, table_name: str) -> ToolDataTable: try: return self.data_tables[table_name] except KeyError: raise exceptions.ObjectNotFound(f"No such data table {table_name}") def _data_table_field(self, table_name: str, field_name: str) -> TabularToolDataField: - out = self._data_table(table_name).get_field(field_name) + out = self._tabular_data_table(table_name).get_field(field_name) if out is None: raise exceptions.ObjectNotFound(f"No such field {field_name} in data table {table_name}.") return out diff --git a/lib/galaxy/tools/data/__init__.py b/lib/galaxy/tools/data/__init__.py index 7cffd2eae27..4bca090c6e8 100644 --- a/lib/galaxy/tools/data/__init__.py +++ b/lib/galaxy/tools/data/__init__.py @@ -18,9 +18,14 @@ import time from glob import glob from tempfile import NamedTemporaryFile from typing import ( + Any, BinaryIO, + Dict, List, Optional, + Set, + Type, + Union, ) import refgenconf @@ -56,12 +61,12 @@ class ToolDataPathFiles: self.update_time = 0 @property - def tool_data_path_files(self): + def tool_data_path_files(self) -> Set[str]: if time.time() - self.update_time > 1: self.update_files() return self._tool_data_path_files - def update_files(self): + def update_files(self) -> None: try: content = os.walk(self.tool_data_path) self._tool_data_path_files = set( @@ -80,7 +85,7 @@ class ToolDataPathFiles: log.exception("Failed to update _tool_data_path_files") self._tool_data_path_files = set() - def exists(self, path): + def exists(self, path: str) -> bool: path = os.path.abspath(path) if path in self.tool_data_path_files: return True @@ -88,11 +93,20 @@ class ToolDataPathFiles: return os.path.exists(path) +ConfigFilesT = Union[str, os.PathLike, List[Union[str, os.PathLike]]] + + class ToolDataTableManager(Dictifiable): """Manages a collection of tool data tables""" + data_tables: Dict[str, "ToolDataTable"] + def __init__( - self, tool_data_path, config_filename=None, tool_data_table_config_path_set=None, other_config_dict=None + self, + tool_data_path: str, + config_filename: Optional[ConfigFilesT] = None, + tool_data_table_config_path_set=None, + other_config_dict=None, ): self.tool_data_path = tool_data_path # This stores all defined data table entries from both the tool_data_table_conf.xml file and the shed_tool_data_table_conf.xml file @@ -110,41 +124,43 @@ class ToolDataTableManager(Dictifiable): data_tables = [ToolDataEntry(**table.to_dict()) for table in self.data_tables.values()] return ToolDataEntryList.construct(__root__=data_tables) - def __getitem__(self, key): + def __getitem__(self, key: str): return self.data_tables.__getitem__(key) - def __setitem__(self, key, value): + def __setitem__(self, key: str, value): return self.data_tables.__setitem__(key, value) - def __contains__(self, key): + def __contains__(self, key: str): return self.data_tables.__contains__(key) - def get(self, name, default=None): + def get(self, name: str, default=None): try: return self[name] except KeyError: return default - def set(self, name, value): + def set(self, name: str, value): self[name] = value - def get_tables(self): + def get_tables(self) -> Dict[str, "ToolDataTable"]: return self.data_tables - def to_dict(self): + def to_dict(self, view: str = "collection", value_mapper=None): return {name: data_table.to_dict(view="export") for name, data_table in self.data_tables.items()} - def to_json(self, path): + def to_json(self, path: Union[str, os.PathLike]) -> None: with open(path, "w") as out: out.write(json.dumps(self.to_dict())) @classmethod - def from_dict(cls, d): + def from_dict(cls, d: Dict[str, Any]): tdtm = cls.__new__(cls) tdtm.data_tables = {name: ToolDataTable.from_dict(data) for name, data in d.items()} return tdtm - def load_from_config_file(self, config_filename, tool_data_path, from_shed_config=False): + def load_from_config_file( + self, config_filename: ConfigFilesT, tool_data_path: Union[str, os.PathLike], from_shed_config: bool = False + ): """ This method is called under 3 conditions: @@ -155,9 +171,12 @@ class ToolDataTableManager(Dictifiable): Galaxy instance. In this case, we have 2 entry types to handle, files whose root tag is , for example: """ table_elems = [] + config_filenames: List[Union[str, os.PathLike]] if not isinstance(config_filename, list): - config_filename = [config_filename] - for filename in config_filename: + config_filenames = [config_filename] + else: + config_filenames = config_filename + for filename in config_filenames: tree = util.parse_xml(filename) root = tree.getroot() for table_elem in root.findall("table"): @@ -297,8 +316,9 @@ class ToolDataTableManager(Dictifiable): return list(table_names) -class ToolDataTable: +class ToolDataTable(Dictifiable): type_key: str + data: List @classmethod def from_elem( @@ -409,7 +429,7 @@ class ToolDataTable: return self._update_version() -class TabularToolDataTable(ToolDataTable, Dictifiable): +class TabularToolDataTable(ToolDataTable): """ Data stored in a tabular / separated value format on disk, allows multiple files to be merged but all must have the same column definitions: @@ -1101,4 +1121,5 @@ def expand_here_template(content, here=None): # Registry of tool data types by type_key -tool_data_table_types = {cls.type_key: cls for cls in [TabularToolDataTable, RefgenieToolDataTable]} +tool_data_table_types_list: List[Type[ToolDataTable]] = [TabularToolDataTable, RefgenieToolDataTable] +tool_data_table_types = {cls.type_key: cls for cls in tool_data_table_types_list} diff --git a/lib/galaxy/util/__init__.py b/lib/galaxy/util/__init__.py index aaab5c45b36..2529ac26c7a 100644 --- a/lib/galaxy/util/__init__.py +++ b/lib/galaxy/util/__init__.py @@ -31,12 +31,15 @@ from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from hashlib import md5 from os.path import relpath +<<<<<<< HEAD from pathlib import Path from typing import ( Any, Optional, overload, ) +======= +>>>>>>> 18a6e028d7 (More tool data table typing and testing.) from urllib.parse import ( urlencode, urlparse, @@ -284,7 +287,7 @@ def unique_id(KEY_SIZE=128): return md5(random_bits).hexdigest() -def parse_xml(fname: typing.Union[str, Path], strip_whitespace=True, remove_comments=True): +def parse_xml(fname: typing.Union[str, os.PathLike], strip_whitespace=True, remove_comments=True): """Returns a parsed xml tree""" parser = None if remove_comments and LXML_AVAILABLE: diff --git a/test/unit/app/tools/test_tool_data.py b/test/unit/app/tools/test_tool_data.py index a12741e12a9..b6ef78d781f 100644 --- a/test/unit/app/tools/test_tool_data.py +++ b/test/unit/app/tools/test_tool_data.py @@ -154,3 +154,10 @@ def test_reload_by_name(tdt_manager, tmp_path): def test_merging_tables(merged_tdt_manager): assert len(merged_tdt_manager["testbeta"].data) == 2 + + +def test_to_json(merged_tdt_manager, tmp_path): + json_path = tmp_path / "as_json.json" + assert not json_path.exists() + merged_tdt_manager.to_json(json_path) + assert json_path.exists()